From 29b3db774e3efacb89e98638f6e6bfe9e2bfed26 Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Wed, 9 Mar 2022 12:47:10 +0530 Subject: [PATCH 1/7] [WIP] Added changes for async callback mechanism from Processor to Exporters --- .../exporters/elasticsearch/es_log_exporter.h | 8 ++ .../elasticsearch/src/es_log_exporter.cc | 124 +++++++++++++++++- .../memory/in_memory_span_exporter.h | 14 ++ .../exporters/ostream/log_exporter.h | 8 ++ .../exporters/ostream/span_exporter.h | 5 + exporters/ostream/src/log_exporter.cc | 10 ++ exporters/ostream/src/span_exporter.cc | 8 ++ .../exporters/zipkin/zipkin_exporter.h | 9 ++ exporters/zipkin/src/zipkin_exporter.cc | 8 ++ .../ext/http/client/curl/http_client_curl.h | 10 +- .../ext/http/client/http_client.h | 2 +- .../http/client/nosend/http_client_nosend.h | 2 +- ext/test/http/curl_http_test.cc | 15 +-- sdk/include/opentelemetry/sdk/logs/exporter.h | 11 ++ .../opentelemetry/sdk/trace/exporter.h | 10 ++ sdk/test/logs/batch_log_processor_test.cc | 7 + sdk/test/logs/simple_log_processor_test.cc | 14 ++ sdk/test/trace/batch_span_processor_test.cc | 8 ++ sdk/test/trace/simple_processor_test.cc | 8 ++ 19 files changed, 263 insertions(+), 18 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h index ea58807e96..14118b2f82 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h @@ -89,6 +89,14 @@ class ElasticsearchLogExporter final : public opentelemetry::sdk::logs::LogExpor const opentelemetry::nostd::span> &records) noexcept override; + /** + * + * + */ + void Export( + const opentelemetry::nostd::span> &records, + nostd::function_ref result_callback) noexcept override; + /** * Shutdown this exporter. * @param timeout The maximum time to wait for the shutdown method to return diff --git a/exporters/elasticsearch/src/es_log_exporter.cc b/exporters/elasticsearch/src/es_log_exporter.cc index a5a66ebe01..05aed5c048 100644 --- a/exporters/elasticsearch/src/es_log_exporter.cc +++ b/exporters/elasticsearch/src/es_log_exporter.cc @@ -110,6 +110,82 @@ class ResponseHandler : public http_client::EventHandler bool console_debug_ = false; }; + +/** + * This class handles the async response message from the Elasticsearch request + */ +class AsyncResponseHandler : public http_client::EventHandler +{ +public: + /** + * Creates a response handler, that by default doesn't display to console + */ + AsyncResponseHandler( + std::shared_ptr session, + nostd::function_ref result_callback, + bool console_debug = false) + : console_debug_{console_debug} + , session_{std::move(session)} + , result_callback_{result_callback} {} + + /** + * Automatically called when the response is received + */ + void OnResponse(http_client::Response &response) noexcept override + { + + // Store the body of the request + body_ = std::string(response.GetBody().begin(), response.GetBody().end()); + session_->FinishSession(); + if (body_.find("\"failed\" : 0") == std::string::npos) + { + OTEL_INTERNAL_LOG_ERROR( + "[ES Trace Exporter] Logs were not written to Elasticsearch correctly, response body: " + << body_); + result_callback_(sdk::common::ExportResult::kFailure); + } else { + result_callback_(sdk::common::ExportResult::kSuccess); + } + } + + // Callback method when an http event occurs + void OnEvent(http_client::SessionState state, nostd::string_view reason) noexcept override + { + // If any failure event occurs, release the condition variable to unblock main thread + switch (state) + { + case http_client::SessionState::ConnectFailed: + OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Connection to elasticsearch failed"); + break; + case http_client::SessionState::SendFailed: + OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Request failed to be sent to elasticsearch"); + + break; + case http_client::SessionState::TimedOut: + OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Request to elasticsearch timed out"); + + break; + case http_client::SessionState::NetworkError: + OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Network error to elasticsearch"); + break; + } + result_callback_(sdk::common::ExportResult::kFailure); + } + +private: + // Stores the session object for the request + std::shared_ptr session_; + // Callback to call to on receiving events + nostd::function_ref result_callback_; + + // A string to store the response body + std::string body_ = ""; + + // Whether to print the results from the callback + bool console_debug_ = false; +}; + + ElasticsearchLogExporter::ElasticsearchLogExporter() : options_{ElasticsearchExporterOptions()}, http_client_{new ext::http::client::curl::HttpClient()} @@ -162,8 +238,8 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export( request->SetBody(body_vec); // Send the request - std::unique_ptr handler(new ResponseHandler(options_.console_debug_)); - session->SendRequest(*handler); + auto handler = std::make_shared(options_.console_debug_); + session->SendRequest(handler); // Wait for the response to be received if (options_.console_debug_) @@ -198,6 +274,50 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export( return sdk::common::ExportResult::kSuccess; } +void ElasticsearchLogExporter::Export( + const opentelemetry::nostd::span> &records, + nostd::function_ref result_callback) noexcept +{ + // Return failure if this exporter has been shutdown + if (isShutdown()) + { + OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting " + << records.size() << " log(s) failed, exporter is shutdown"); + return; + } + + // Create a connection to the ElasticSearch instance + auto session = http_client_->CreateSession(options_.host_ + std::to_string(options_.port_)); + auto request = session->CreateRequest(); + + // Populate the request with headers and methods + request->SetUri(options_.index_ + "/_bulk?pretty"); + request->SetMethod(http_client::Method::Post); + request->AddHeader("Content-Type", "application/json"); + request->SetTimeoutMs(std::chrono::milliseconds(1000 * options_.response_timeout_)); + + // Create the request body + std::string body = ""; + for (auto &record : records) + { + // Append {"index":{}} before JSON body, which tells Elasticsearch to write to index specified + // in URI + body += "{\"index\" : {}}\n"; + + // Add the context of the Recordable + auto json_record = std::unique_ptr( + static_cast(record.release())); + body += json_record->GetJSON().dump() + "\n"; + } + std::vector body_vec(body.begin(), body.end()); + request->SetBody(body_vec); + + // Send the request + auto handler = std::make_shared( + session, result_callback, options_.console_debug_); + session->SendRequest(handler); +} + bool ElasticsearchLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept { const std::lock_guard locked(lock_); diff --git a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h index 3e47ccb177..ada527201f 100644 --- a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h +++ b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h @@ -64,6 +64,20 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte return sdk::common::ExportResult::kSuccess; } + /** + * + * + */ + void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept override + { + auto result = Export(spans); + result_callback(result); + + } + /** * @param timeout an optional value containing the timeout of the exporter * note: passing custom timeout values is not currently supported for this exporter diff --git a/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h b/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h index ad1d54a215..2a1f7b6aa2 100644 --- a/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h +++ b/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h @@ -39,6 +39,14 @@ class OStreamLogExporter final : public opentelemetry::sdk::logs::LogExporter const opentelemetry::nostd::span> &records) noexcept override; + /** + * Exports a span of logs sent from the processor asynchronously. + */ + void Export( + const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref result_callback) + noexcept; + /** * Marks the OStream Log Exporter as shut down. */ diff --git a/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h b/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h index 8122b6777a..d23a92f77f 100644 --- a/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h +++ b/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h @@ -38,6 +38,11 @@ class OStreamSpanExporter final : public opentelemetry::sdk::trace::SpanExporter const opentelemetry::nostd::span> &spans) noexcept override; + void Export( + const opentelemetry::nostd::span> &spans, + opentelemetry::nostd::function_ref result_callback) + noexcept override; + bool Shutdown( std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override; diff --git a/exporters/ostream/src/log_exporter.cc b/exporters/ostream/src/log_exporter.cc index ef103bb6b2..cc97073123 100644 --- a/exporters/ostream/src/log_exporter.cc +++ b/exporters/ostream/src/log_exporter.cc @@ -180,6 +180,16 @@ sdk::common::ExportResult OStreamLogExporter::Export( return sdk::common::ExportResult::kSuccess; } +void OStreamLogExporter::Export( + const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref result_callback) + noexcept +{ + // Do not have async support + auto result = Export(records); + result_callback(result); +} + bool OStreamLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept { const std::lock_guard locked(lock_); diff --git a/exporters/ostream/src/span_exporter.cc b/exporters/ostream/src/span_exporter.cc index dea72f57f8..2a97f830c3 100644 --- a/exporters/ostream/src/span_exporter.cc +++ b/exporters/ostream/src/span_exporter.cc @@ -96,6 +96,14 @@ sdk::common::ExportResult OStreamSpanExporter::Export( return sdk::common::ExportResult::kSuccess; } +void OStreamSpanExporter::Export( + const opentelemetry::nostd::span> &spans, + opentelemetry::nostd::function_ref result_callback) noexcept +{ + auto result = Export(spans); + result_callback(result); +} + bool OStreamSpanExporter::Shutdown(std::chrono::microseconds timeout) noexcept { const std::lock_guard locked(lock_); diff --git a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h index ae0e8173f9..e350fb276a 100644 --- a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h +++ b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h @@ -78,6 +78,15 @@ class ZipkinExporter final : public opentelemetry::sdk::trace::SpanExporter const nostd::span> &spans) noexcept override; + /** + * + * + */ + void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept override; + /** * Shut down the exporter. * @param timeout an optional timeout, default to max. diff --git a/exporters/zipkin/src/zipkin_exporter.cc b/exporters/zipkin/src/zipkin_exporter.cc index 240144599f..b73468cd7a 100644 --- a/exporters/zipkin/src/zipkin_exporter.cc +++ b/exporters/zipkin/src/zipkin_exporter.cc @@ -93,6 +93,14 @@ sdk::common::ExportResult ZipkinExporter::Export( return sdk::common::ExportResult::kSuccess; } +void ZipkinExporter::Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept +{ + +} + void ZipkinExporter::InitializeLocalEndpoint() { if (options_.service_name.length()) diff --git a/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h b/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h index 9f2f05f3f0..e6f9c68017 100644 --- a/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h +++ b/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h @@ -143,19 +143,19 @@ class Session : public opentelemetry::ext::http::client::Session } virtual void SendRequest( - opentelemetry::ext::http::client::EventHandler &callback) noexcept override + std::shared_ptr callback) noexcept override { is_session_active_ = true; std::string url = host_ + std::string(http_request_->uri_); - auto callback_ptr = &callback; + auto callback_ptr = callback.get(); curl_operation_.reset(new HttpOperation( http_request_->method_, url, callback_ptr, RequestMode::Async, http_request_->headers_, http_request_->body_, false, http_request_->timeout_ms_)); - curl_operation_->SendAsync([this, callback_ptr](HttpOperation &operation) { + curl_operation_->SendAsync([this, callback](HttpOperation &operation) { if (operation.WasAborted()) { // Manually cancelled - callback_ptr->OnEvent(opentelemetry::ext::http::client::SessionState::Cancelled, ""); + callback->OnEvent(opentelemetry::ext::http::client::SessionState::Cancelled, ""); } if (operation.GetResponseCode() >= CURL_LAST) @@ -165,7 +165,7 @@ class Session : public opentelemetry::ext::http::client::Session response->headers_ = operation.GetResponseHeaders(); response->body_ = operation.GetResponseBody(); response->status_code_ = operation.GetResponseCode(); - callback_ptr->OnResponse(*response); + callback->OnResponse(*response); } is_session_active_ = false; }); diff --git a/ext/include/opentelemetry/ext/http/client/http_client.h b/ext/include/opentelemetry/ext/http/client/http_client.h index 308335e492..e939962653 100644 --- a/ext/include/opentelemetry/ext/http/client/http_client.h +++ b/ext/include/opentelemetry/ext/http/client/http_client.h @@ -212,7 +212,7 @@ class Session public: virtual std::shared_ptr CreateRequest() noexcept = 0; - virtual void SendRequest(EventHandler &) noexcept = 0; + virtual void SendRequest(std::shared_ptr) noexcept = 0; virtual bool IsSessionActive() noexcept = 0; diff --git a/ext/include/opentelemetry/ext/http/client/nosend/http_client_nosend.h b/ext/include/opentelemetry/ext/http/client/nosend/http_client_nosend.h index 02433d75ce..bc68592743 100644 --- a/ext/include/opentelemetry/ext/http/client/nosend/http_client_nosend.h +++ b/ext/include/opentelemetry/ext/http/client/nosend/http_client_nosend.h @@ -121,7 +121,7 @@ class Session : public opentelemetry::ext::http::client::Session MOCK_METHOD(void, SendRequest, - (opentelemetry::ext::http::client::EventHandler &), + (std::shared_ptr), (noexcept, override)); virtual bool CancelSession() noexcept override; diff --git a/ext/test/http/curl_http_test.cc b/ext/test/http/curl_http_test.cc index f8d248bae4..3b48d2aa1e 100644 --- a/ext/test/http/curl_http_test.cc +++ b/ext/test/http/curl_http_test.cc @@ -196,12 +196,11 @@ TEST_F(BasicCurlHttpTests, SendGetRequest) auto session = session_manager->CreateSession("http://127.0.0.1:19000"); auto request = session->CreateRequest(); request->SetUri("get/"); - GetEventHandler *handler = new GetEventHandler(); - session->SendRequest(*handler); + auto handler = std::make_shared(); + session->SendRequest(handler); ASSERT_TRUE(waitForRequests(30, 1)); session->FinishSession(); ASSERT_TRUE(handler->is_called_); - delete handler; } TEST_F(BasicCurlHttpTests, SendPostRequest) @@ -219,8 +218,8 @@ TEST_F(BasicCurlHttpTests, SendPostRequest) http_client::Body body = {b, b + strlen(b)}; request->SetBody(body); request->AddHeader("Content-Type", "text/plain"); - PostEventHandler *handler = new PostEventHandler(); - session->SendRequest(*handler); + auto handler = std::make_shared(); + session->SendRequest(handler); ASSERT_TRUE(waitForRequests(30, 1)); session->FinishSession(); ASSERT_TRUE(handler->is_called_); @@ -228,7 +227,6 @@ TEST_F(BasicCurlHttpTests, SendPostRequest) session_manager->CancelAllSessions(); session_manager->FinishAllSessions(); - delete handler; } TEST_F(BasicCurlHttpTests, RequestTimeout) @@ -240,11 +238,10 @@ TEST_F(BasicCurlHttpTests, RequestTimeout) auto session = session_manager->CreateSession("222.222.222.200:19000"); // Non Existing address auto request = session->CreateRequest(); request->SetUri("get/"); - GetEventHandler *handler = new GetEventHandler(); - session->SendRequest(*handler); + auto handler = std::make_shared(); + session->SendRequest(handler); session->FinishSession(); ASSERT_FALSE(handler->is_called_); - delete handler; } TEST_F(BasicCurlHttpTests, CurlHttpOperations) diff --git a/sdk/include/opentelemetry/sdk/logs/exporter.h b/sdk/include/opentelemetry/sdk/logs/exporter.h index 85c58e9f12..829c9e1432 100644 --- a/sdk/include/opentelemetry/sdk/logs/exporter.h +++ b/sdk/include/opentelemetry/sdk/logs/exporter.h @@ -46,6 +46,17 @@ class LogExporter virtual sdk::common::ExportResult Export( const nostd::span> &records) noexcept = 0; + + /** + * Exports the batch of log records to their export destination + * + * + * + */ + virtual void Export( + const nostd::span> &records, + nostd::function_ref result_callback) noexcept = 0; + /** * Marks the exporter as ShutDown and cleans up any resources as required. * Shutdown should be called only once for each Exporter instance. diff --git a/sdk/include/opentelemetry/sdk/trace/exporter.h b/sdk/include/opentelemetry/sdk/trace/exporter.h index 5826b5f454..4da783b826 100644 --- a/sdk/include/opentelemetry/sdk/trace/exporter.h +++ b/sdk/include/opentelemetry/sdk/trace/exporter.h @@ -42,6 +42,16 @@ class SpanExporter const nostd::span> &spans) noexcept = 0; + /** + * Exports a batch of span recordables. + * + * + * + */ + virtual void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept = 0; + /** * Shut down the exporter. * @param timeout an optional timeout. diff --git a/sdk/test/logs/batch_log_processor_test.cc b/sdk/test/logs/batch_log_processor_test.cc index df503cb2aa..2a8dc13bf3 100644 --- a/sdk/test/logs/batch_log_processor_test.cc +++ b/sdk/test/logs/batch_log_processor_test.cc @@ -55,6 +55,13 @@ class MockLogExporter final : public LogExporter return ExportResult::kSuccess; } + void Export(const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref result_callback) noexcept override + { + auto result = Export(records); + result_callback(result); + } + // toggles the boolean flag marking this exporter as shut down bool Shutdown( std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override diff --git a/sdk/test/logs/simple_log_processor_test.cc b/sdk/test/logs/simple_log_processor_test.cc index 0bb6ba2667..2a86910c94 100644 --- a/sdk/test/logs/simple_log_processor_test.cc +++ b/sdk/test/logs/simple_log_processor_test.cc @@ -53,6 +53,14 @@ class TestExporter final : public LogExporter return ExportResult::kSuccess; } + // Dummy Async Export implementation + void Export(const nostd::span> &records, + nostd::function_ref result_callback) noexcept override + { + auto result = Export(records); + result_callback(result); + } + // Increment the shutdown counter everytime this method is called bool Shutdown(std::chrono::microseconds timeout) noexcept override { @@ -137,6 +145,12 @@ class FailShutDownExporter final : public LogExporter return ExportResult::kSuccess; } + void Export(const nostd::span> &records, + nostd::function_ref result_callback) noexcept override + { + result_callback(ExportResult::kSuccess); + } + bool Shutdown(std::chrono::microseconds timeout) noexcept override { return false; } }; diff --git a/sdk/test/trace/batch_span_processor_test.cc b/sdk/test/trace/batch_span_processor_test.cc index 0e6f9c35aa..a445315aa2 100644 --- a/sdk/test/trace/batch_span_processor_test.cc +++ b/sdk/test/trace/batch_span_processor_test.cc @@ -56,6 +56,14 @@ class MockSpanExporter final : public sdk::trace::SpanExporter return sdk::common::ExportResult::kSuccess; } + void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept override + { + auto result = Export(spans); + result_callback(result); + } + bool Shutdown( std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override { diff --git a/sdk/test/trace/simple_processor_test.cc b/sdk/test/trace/simple_processor_test.cc index 9398b922a5..b46339ebad 100644 --- a/sdk/test/trace/simple_processor_test.cc +++ b/sdk/test/trace/simple_processor_test.cc @@ -51,6 +51,14 @@ class RecordShutdownExporter final : public SpanExporter return ExportResult::kSuccess; } + void Export( + const opentelemetry::nostd::span> &spans, + opentelemetry::nostd::function_ref result_callback) + noexcept override + { + result_callback(ExportResult::kSuccess); + } + bool Shutdown( std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override { From a746a986d06495ec13ca2ae7f7a9608fdecf1331 Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Wed, 9 Mar 2022 16:32:38 +0530 Subject: [PATCH 2/7] Added otlp files --- exporters/elasticsearch/src/es_log_exporter.cc | 9 ++++++++- .../opentelemetry/exporters/jaeger/jaeger_exporter.h | 11 +++++++++++ exporters/jaeger/src/jaeger_exporter.cc | 8 ++++++++ .../opentelemetry/exporters/otlp/otlp_grpc_exporter.h | 9 +++++++++ .../exporters/otlp/otlp_grpc_log_exporter.h | 10 ++++++++++ .../opentelemetry/exporters/otlp/otlp_http_exporter.h | 10 ++++++++++ .../exporters/otlp/otlp_http_log_exporter.h | 10 ++++++++++ exporters/otlp/src/otlp_grpc_exporter.cc | 8 ++++++++ exporters/otlp/src/otlp_grpc_log_exporter.cc | 7 +++++++ exporters/otlp/src/otlp_http_client.cc | 4 ++-- exporters/otlp/src/otlp_http_exporter.cc | 7 +++++++ exporters/otlp/src/otlp_http_log_exporter.cc | 7 +++++++ exporters/otlp/test/otlp_http_exporter_test.cc | 8 ++++---- exporters/otlp/test/otlp_http_log_exporter_test.cc | 8 ++++---- sdk/include/opentelemetry/sdk/trace/exporter.h | 7 +++---- 15 files changed, 108 insertions(+), 15 deletions(-) diff --git a/exporters/elasticsearch/src/es_log_exporter.cc b/exporters/elasticsearch/src/es_log_exporter.cc index 05aed5c048..4c8969ae33 100644 --- a/exporters/elasticsearch/src/es_log_exporter.cc +++ b/exporters/elasticsearch/src/es_log_exporter.cc @@ -128,6 +128,14 @@ class AsyncResponseHandler : public http_client::EventHandler , session_{std::move(session)} , result_callback_{result_callback} {} + /** + * Cleans up the session in the destructor. + */ + ~AsyncResponseHandler() + { + session_->FinishSession(); + } + /** * Automatically called when the response is received */ @@ -136,7 +144,6 @@ class AsyncResponseHandler : public http_client::EventHandler // Store the body of the request body_ = std::string(response.GetBody().begin(), response.GetBody().end()); - session_->FinishSession(); if (body_.find("\"failed\" : 0") == std::string::npos) { OTEL_INTERNAL_LOG_ERROR( diff --git a/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h b/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h index 284bab2cab..c2e17b97ff 100644 --- a/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h +++ b/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h @@ -61,6 +61,17 @@ class JaegerExporter final : public opentelemetry::sdk::trace::SpanExporter const nostd::span> &spans) noexcept override; + /** + * Exports a batch of span recordables asynchronously. + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument + */ + void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept override; + + /** * Shutdown the exporter. * @param timeout an option timeout, default to max. diff --git a/exporters/jaeger/src/jaeger_exporter.cc b/exporters/jaeger/src/jaeger_exporter.cc index c07f2f0100..abe1610b5f 100644 --- a/exporters/jaeger/src/jaeger_exporter.cc +++ b/exporters/jaeger/src/jaeger_exporter.cc @@ -70,6 +70,14 @@ sdk_common::ExportResult JaegerExporter::Export( return sdk_common::ExportResult::kSuccess; } +void JaegerExporter::Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept +{ + +} + void JaegerExporter::InitializeEndpoint() { if (options_.transport_format == TransportFormat::kThriftUdpCompact) diff --git a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h index a28e6fca85..722e525b04 100644 --- a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h +++ b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h @@ -52,6 +52,15 @@ class OtlpGrpcExporter final : public opentelemetry::sdk::trace::SpanExporter sdk::common::ExportResult Export( const nostd::span> &spans) noexcept override; + /** + * Exports a batch of span recordables asynchronously. + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument + */ + virtual void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept override; + /** * Shut down the exporter. * @param timeout an optional timeout, the default timeout of 0 means that no diff --git a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_exporter.h b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_exporter.h index a8aeda85b8..5b7f79e13b 100644 --- a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_exporter.h +++ b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_exporter.h @@ -55,6 +55,16 @@ class OtlpGrpcLogExporter : public opentelemetry::sdk::logs::LogExporter const nostd::span> &records) noexcept override; + /** + * Exports a vector of log records asynchronously. + * @param records A list of log records. + * @param result_callback callback function accepting ExportResult as argument + */ + virtual void Export( + const nostd::span> &records, + nostd::function_ref result_callback) noexcept + override; + /** * Shutdown this exporter. * @param timeout The maximum time to wait for the shutdown method to return. diff --git a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h index 3e6a521194..554a5c9fe1 100644 --- a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h +++ b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h @@ -82,6 +82,16 @@ class OtlpHttpExporter final : public opentelemetry::sdk::trace::SpanExporter const nostd::span> &spans) noexcept override; + /** + * Exports a batch of span recordables asynchronously. + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument + */ + virtual void Export( + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept + override; + /** * Shut down the exporter. * @param timeout an optional timeout, the default timeout of 0 means that no diff --git a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h index d330e62be4..195a22c867 100644 --- a/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h +++ b/exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h @@ -83,6 +83,16 @@ class OtlpHttpLogExporter final : public opentelemetry::sdk::logs::LogExporter const nostd::span> &records) noexcept override; + /** + * Exports a vector of log records asynchronously. + * @param records A list of log records. + * @param result_callback callback function accepting ExportResult as argument + */ + virtual void Export( + const nostd::span> &records, + nostd::function_ref result_callback) noexcept + override; + /** * Shutdown this exporter. * @param timeout The maximum time to wait for the shutdown method to return diff --git a/exporters/otlp/src/otlp_grpc_exporter.cc b/exporters/otlp/src/otlp_grpc_exporter.cc index 32f4a60a52..686c4df6ed 100644 --- a/exporters/otlp/src/otlp_grpc_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_exporter.cc @@ -143,6 +143,14 @@ sdk::common::ExportResult OtlpGrpcExporter::Export( return sdk::common::ExportResult::kSuccess; } +void OtlpGrpcExporter::Export( + const nostd::span> &spans, + nostd::function_ref result_callback) + noexcept +{ + +} + bool OtlpGrpcExporter::Shutdown(std::chrono::microseconds timeout) noexcept { const std::lock_guard locked(lock_); diff --git a/exporters/otlp/src/otlp_grpc_log_exporter.cc b/exporters/otlp/src/otlp_grpc_log_exporter.cc index 38bfb0a5bb..48d64157f5 100644 --- a/exporters/otlp/src/otlp_grpc_log_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_log_exporter.cc @@ -161,6 +161,13 @@ opentelemetry::sdk::common::ExportResult OtlpGrpcLogExporter::Export( return sdk::common::ExportResult::kSuccess; } +void OtlpGrpcLogExporter::Export( + const nostd::span> &logs, + nostd::function_ref result_callback) noexcept +{ + +} + bool OtlpGrpcLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept { const std::lock_guard locked(lock_); diff --git a/exporters/otlp/src/otlp_http_client.cc b/exporters/otlp/src/otlp_http_client.cc index 544f74ca7c..45b1599a6a 100644 --- a/exporters/otlp/src/otlp_http_client.cc +++ b/exporters/otlp/src/otlp_http_client.cc @@ -668,8 +668,8 @@ opentelemetry::sdk::common::ExportResult OtlpHttpClient::Export( request->ReplaceHeader("Content-Type", content_type); // Send the request - std::unique_ptr handler(new ResponseHandler(options_.console_debug)); - session->SendRequest(*handler); + auto handler = std::make_shared(options_.console_debug); + session->SendRequest(handler); // Wait for the response to be received if (options_.console_debug) diff --git a/exporters/otlp/src/otlp_http_exporter.cc b/exporters/otlp/src/otlp_http_exporter.cc index 92155dd00d..fa603988f7 100644 --- a/exporters/otlp/src/otlp_http_exporter.cc +++ b/exporters/otlp/src/otlp_http_exporter.cc @@ -56,6 +56,13 @@ opentelemetry::sdk::common::ExportResult OtlpHttpExporter::Export( return http_client_->Export(service_request); } +void OtlpHttpExporter::Export( + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept +{ + +} + bool OtlpHttpExporter::Shutdown(std::chrono::microseconds timeout) noexcept { return http_client_->Shutdown(timeout); diff --git a/exporters/otlp/src/otlp_http_log_exporter.cc b/exporters/otlp/src/otlp_http_log_exporter.cc index 436c77beaa..198ea47b1d 100644 --- a/exporters/otlp/src/otlp_http_log_exporter.cc +++ b/exporters/otlp/src/otlp_http_log_exporter.cc @@ -57,6 +57,13 @@ opentelemetry::sdk::common::ExportResult OtlpHttpLogExporter::Export( return http_client_->Export(service_request); } +void OtlpHttpLogExporter::Export( + const nostd::span> &logs, + nostd::function_ref result_callback) noexcept +{ + +} + bool OtlpHttpLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept { return http_client_->Shutdown(timeout); diff --git a/exporters/otlp/test/otlp_http_exporter_test.cc b/exporters/otlp/test/otlp_http_exporter_test.cc index ef0b5a509e..ef9f824733 100644 --- a/exporters/otlp/test/otlp_http_exporter_test.cc +++ b/exporters/otlp/test/otlp_http_exporter_test.cc @@ -139,7 +139,7 @@ TEST_F(OtlpHttpExporterTestPeer, ExportJsonIntegrationTest) std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) .WillOnce([&mock_session, - report_trace_id](opentelemetry::ext::http::client::EventHandler &callback) { + report_trace_id](std::shared_ptr callback) { auto check_json = nlohmann::json::parse(mock_session->GetRequest()->body_, nullptr, false); auto resource_span = *check_json["resource_spans"].begin(); auto instrumentation_library_span = *resource_span["instrumentation_library_spans"].begin(); @@ -155,7 +155,7 @@ TEST_F(OtlpHttpExporterTestPeer, ExportJsonIntegrationTest) } // let the otlp_http_client to continue http_client::nosend::Response response; - callback.OnResponse(response); + callback->OnResponse(response); }); child_span->End(); @@ -218,7 +218,7 @@ TEST_F(OtlpHttpExporterTestPeer, ExportBinaryIntegrationTest) std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) .WillOnce([&mock_session, - report_trace_id](opentelemetry::ext::http::client::EventHandler &callback) { + report_trace_id](std::shared_ptr callback) { opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest request_body; request_body.ParseFromArray(&mock_session->GetRequest()->body_[0], static_cast(mock_session->GetRequest()->body_.size())); @@ -234,7 +234,7 @@ TEST_F(OtlpHttpExporterTestPeer, ExportBinaryIntegrationTest) } // let the otlp_http_client to continue http_client::nosend::Response response; - callback.OnResponse(response); + callback->OnResponse(response); }); child_span->End(); diff --git a/exporters/otlp/test/otlp_http_log_exporter_test.cc b/exporters/otlp/test/otlp_http_log_exporter_test.cc index ffd1a9a0f3..c7f52db214 100644 --- a/exporters/otlp/test/otlp_http_log_exporter_test.cc +++ b/exporters/otlp/test/otlp_http_log_exporter_test.cc @@ -149,7 +149,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportJsonIntegrationTest) std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) .WillOnce([&mock_session, report_trace_id, - report_span_id](opentelemetry::ext::http::client::EventHandler &callback) { + report_span_id](std::shared_ptr callback) { auto check_json = nlohmann::json::parse(mock_session->GetRequest()->body_, nullptr, false); auto resource_logs = *check_json["resource_logs"].begin(); auto instrumentation_library_span = *resource_logs["instrumentation_library_logs"].begin(); @@ -169,7 +169,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportJsonIntegrationTest) } // let the otlp_http_client to continue http_client::nosend::Response response; - callback.OnResponse(response); + callback->OnResponse(response); }); } @@ -233,7 +233,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportBinaryIntegrationTest) std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) .WillOnce([&mock_session, report_trace_id, - report_span_id](opentelemetry::ext::http::client::EventHandler &callback) { + report_span_id](std::shared_ptr callback) { opentelemetry::proto::collector::logs::v1::ExportLogsServiceRequest request_body; request_body.ParseFromArray(&mock_session->GetRequest()->body_[0], static_cast(mock_session->GetRequest()->body_.size())); @@ -254,7 +254,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportBinaryIntegrationTest) } ASSERT_TRUE(check_service_name); http_client::nosend::Response response; - callback.OnResponse(response); + callback->OnResponse(response); }); } diff --git a/sdk/include/opentelemetry/sdk/trace/exporter.h b/sdk/include/opentelemetry/sdk/trace/exporter.h index 4da783b826..8078b23e48 100644 --- a/sdk/include/opentelemetry/sdk/trace/exporter.h +++ b/sdk/include/opentelemetry/sdk/trace/exporter.h @@ -43,10 +43,9 @@ class SpanExporter &spans) noexcept = 0; /** - * Exports a batch of span recordables. - * - * - * + * Exports a batch of span recordables asynchronously. + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument */ virtual void Export( const nostd::span> &spans, From 7e8ac323aab012f74d6c7bb439f45d7e98c0d18a Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Wed, 9 Mar 2022 18:21:42 +0530 Subject: [PATCH 3/7] Added batch processor changes --- .../sdk/logs/batch_log_processor.h | 4 ++- .../sdk/trace/batch_span_processor.h | 7 +++++ sdk/src/logs/batch_log_processor.cc | 31 +++++++++++++------ sdk/src/trace/batch_span_processor.cc | 27 +++++++++++----- 4 files changed, 51 insertions(+), 18 deletions(-) diff --git a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h index 1b6d443c8a..fa462cb519 100644 --- a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h @@ -41,7 +41,8 @@ class BatchLogProcessor : public LogProcessor std::unique_ptr &&exporter, const size_t max_queue_size = 2048, const std::chrono::milliseconds scheduled_delay_millis = std::chrono::milliseconds(5000), - const size_t max_export_batch_size = 512); + const size_t max_export_batch_size = 512, + const bool is_export_async = false); /** Makes a new recordable **/ std::unique_ptr MakeRecordable() noexcept override; @@ -105,6 +106,7 @@ class BatchLogProcessor : public LogProcessor const size_t max_queue_size_; const std::chrono::milliseconds scheduled_delay_millis_; const size_t max_export_batch_size_; + const bool is_export_async_; /* Synchronization primitives */ std::condition_variable cv_, force_flush_cv_; diff --git a/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h b/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h index d25ff2d950..b134e877ef 100644 --- a/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h +++ b/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h @@ -37,6 +37,12 @@ struct BatchSpanProcessorOptions * equal to max_queue_size. */ size_t max_export_batch_size = 512; + + /** + * Determines whether the export happens asynchronously. + * Default implementation is synchronous. + */ + bool is_export_async = false; }; /** @@ -136,6 +142,7 @@ class BatchSpanProcessor : public SpanProcessor const size_t max_queue_size_; const std::chrono::milliseconds schedule_delay_millis_; const size_t max_export_batch_size_; + const bool is_export_async_; /* Synchronization primitives */ std::condition_variable cv_, force_flush_cv_; diff --git a/sdk/src/logs/batch_log_processor.cc b/sdk/src/logs/batch_log_processor.cc index 9b20705b0a..880e9bbb2f 100644 --- a/sdk/src/logs/batch_log_processor.cc +++ b/sdk/src/logs/batch_log_processor.cc @@ -16,12 +16,14 @@ namespace logs BatchLogProcessor::BatchLogProcessor(std::unique_ptr &&exporter, const size_t max_queue_size, const std::chrono::milliseconds scheduled_delay_millis, - const size_t max_export_batch_size) + const size_t max_export_batch_size, + const bool is_export_async) : exporter_(std::move(exporter)), max_queue_size_(max_queue_size), scheduled_delay_millis_(scheduled_delay_millis), max_export_batch_size_(max_export_batch_size), buffer_(max_queue_size_), + is_export_async_(is_export_async), worker_thread_(&BatchLogProcessor::DoBackgroundWork, this) {} @@ -151,19 +153,28 @@ void BatchLogProcessor::Export(const bool was_force_flush_called) return true; }); }); + if (is_export_async_ == false || was_force_flush_called == true) { + exporter_->Export( + nostd::span>(records_arr.data(), records_arr.size())); - exporter_->Export( - nostd::span>(records_arr.data(), records_arr.size())); - - // Notify the main thread in case this export was the result of a force flush. - if (was_force_flush_called == true) - { - is_force_flush_notified_ = true; - while (is_force_flush_notified_.load() == true) + // Notify the main thread in case this export was the result of a force flush. + if (was_force_flush_called == true) { - force_flush_cv_.notify_one(); + is_force_flush_notified_ = true; + while (is_force_flush_notified_.load() == true) + { + force_flush_cv_.notify_one(); + } } } + else { + exporter_->Export( + nostd::span>(records_arr.data(), records_arr.size()), + [](sdk::common::ExportResult result) { + // TODO: Print result + return true; + }); + } } void BatchLogProcessor::DrainQueue() diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index 0ab042b9ab..5c9801a441 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -21,6 +21,7 @@ BatchSpanProcessor::BatchSpanProcessor(std::unique_ptr &&exporter, schedule_delay_millis_(options.schedule_delay_millis), max_export_batch_size_(options.max_export_batch_size), buffer_(max_queue_size_), + is_export_async_(options.is_export_async), worker_thread_(&BatchSpanProcessor::DoBackgroundWork, this) {} @@ -157,17 +158,29 @@ void BatchSpanProcessor::Export(const bool was_force_flush_called) }); }); - exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size())); + /* Call the sync Export when force flush was called, even if + is_export_async_ is true. + */ + if (is_export_async_ == false || was_force_flush_called == true) { + exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size())); - // Notify the main thread in case this export was the result of a force flush. - if (was_force_flush_called == true) - { - is_force_flush_notified_ = true; - while (is_force_flush_notified_.load() == true) + // Notify the main thread in case this export was the result of a force flush. + if (was_force_flush_called == true) { - force_flush_cv_.notify_one(); + is_force_flush_notified_ = true; + while (is_force_flush_notified_.load() == true) + { + force_flush_cv_.notify_one(); + } } } + else { + exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size()), + [](sdk::common::ExportResult result) { + // TODO: Print result + return true; + }); + } } void BatchSpanProcessor::DrainQueue() From c58f0cdb24ca87f426d70d21e4f20afaf042b4a0 Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Thu, 10 Mar 2022 17:37:30 +0530 Subject: [PATCH 4/7] Incorporated first level of review comments --- exporters/jaeger/src/jaeger_exporter.cc | 3 +++ .../exporters/memory/in_memory_span_exporter.h | 6 +++--- exporters/otlp/src/otlp_grpc_exporter.cc | 3 +++ exporters/otlp/src/otlp_grpc_log_exporter.cc | 4 +++- exporters/otlp/src/otlp_http_exporter.cc | 7 ++++++- exporters/otlp/src/otlp_http_log_exporter.cc | 6 +++++- exporters/zipkin/src/zipkin_exporter.cc | 4 +++- ext/test/w3c_tracecontext_test/main.cc | 4 ++-- 8 files changed, 28 insertions(+), 9 deletions(-) diff --git a/exporters/jaeger/src/jaeger_exporter.cc b/exporters/jaeger/src/jaeger_exporter.cc index abe1610b5f..60d8a80db9 100644 --- a/exporters/jaeger/src/jaeger_exporter.cc +++ b/exporters/jaeger/src/jaeger_exporter.cc @@ -75,6 +75,9 @@ void JaegerExporter::Export( nostd::function_ref result_callback) noexcept { + OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); + auto status = Export(spans); + result_callback(status); } diff --git a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h index ada527201f..565a118a51 100644 --- a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h +++ b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h @@ -73,9 +73,9 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte nostd::function_ref result_callback) noexcept override { - auto result = Export(spans); - result_callback(result); - + OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); + auto status = Export(spans); + result_callback(status); } /** diff --git a/exporters/otlp/src/otlp_grpc_exporter.cc b/exporters/otlp/src/otlp_grpc_exporter.cc index 686c4df6ed..b06e788388 100644 --- a/exporters/otlp/src/otlp_grpc_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_exporter.cc @@ -148,6 +148,9 @@ void OtlpGrpcExporter::Export( nostd::function_ref result_callback) noexcept { + OTEL_INTERNAL_LOG_WARN("[OTLP TRACE GRPC Exporter] async not supported. Making sync interface call"); + auto status = Export(spans); + result_callback(status); } diff --git a/exporters/otlp/src/otlp_grpc_log_exporter.cc b/exporters/otlp/src/otlp_grpc_log_exporter.cc index 48d64157f5..91e2a56e85 100644 --- a/exporters/otlp/src/otlp_grpc_log_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_log_exporter.cc @@ -165,7 +165,9 @@ void OtlpGrpcLogExporter::Export( const nostd::span> &logs, nostd::function_ref result_callback) noexcept { - + OTEL_INTERNAL_LOG_WARN("[OTLP LOG GRPC Exporter] async not supported. Making sync interface call"); + auto status = Export(logs); + result_callback(status); } bool OtlpGrpcLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept diff --git a/exporters/otlp/src/otlp_http_exporter.cc b/exporters/otlp/src/otlp_http_exporter.cc index fa603988f7..452fa89094 100644 --- a/exporters/otlp/src/otlp_http_exporter.cc +++ b/exporters/otlp/src/otlp_http_exporter.cc @@ -11,6 +11,9 @@ #include "opentelemetry/exporters/otlp/protobuf_include_suffix.h" +#include "opentelemetry/sdk/common/global_log_handler.h" + + namespace nostd = opentelemetry::nostd; OPENTELEMETRY_BEGIN_NAMESPACE @@ -60,7 +63,9 @@ void OtlpHttpExporter::Export( const nostd::span> &spans, nostd::function_ref result_callback) noexcept { - + OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); + auto status = Export(spans); + result_callback(status); } bool OtlpHttpExporter::Shutdown(std::chrono::microseconds timeout) noexcept diff --git a/exporters/otlp/src/otlp_http_log_exporter.cc b/exporters/otlp/src/otlp_http_log_exporter.cc index 198ea47b1d..734d6ecc48 100644 --- a/exporters/otlp/src/otlp_http_log_exporter.cc +++ b/exporters/otlp/src/otlp_http_log_exporter.cc @@ -13,6 +13,8 @@ # include "opentelemetry/exporters/otlp/protobuf_include_suffix.h" +# include "opentelemetry/sdk/common/global_log_handler.h" + namespace nostd = opentelemetry::nostd; OPENTELEMETRY_BEGIN_NAMESPACE @@ -61,7 +63,9 @@ void OtlpHttpLogExporter::Export( const nostd::span> &logs, nostd::function_ref result_callback) noexcept { - + OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); + auto status = Export(logs); + result_callback(status); } bool OtlpHttpLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept diff --git a/exporters/zipkin/src/zipkin_exporter.cc b/exporters/zipkin/src/zipkin_exporter.cc index b73468cd7a..e3341945c8 100644 --- a/exporters/zipkin/src/zipkin_exporter.cc +++ b/exporters/zipkin/src/zipkin_exporter.cc @@ -98,7 +98,9 @@ void ZipkinExporter::Export( nostd::function_ref result_callback) noexcept { - + OTEL_INTERNAL_LOG_WARN("[ZIPKIN EXPORTER] async not supported. Making sync interface call"); + auto status = Export(spans); + result_callback(status); } void ZipkinExporter::InitializeLocalEndpoint() diff --git a/ext/test/w3c_tracecontext_test/main.cc b/ext/test/w3c_tracecontext_test/main.cc index 79aa4c9169..ca54475540 100644 --- a/ext/test/w3c_tracecontext_test/main.cc +++ b/ext/test/w3c_tracecontext_test/main.cc @@ -100,7 +100,7 @@ class NoopEventHandler : public http_client::EventHandler // Sends an HTTP POST request to the given url, with the given body. void send_request(curl::HttpClient &client, const std::string &url, const std::string &body) { - static std::unique_ptr handler(new NoopEventHandler()); + static std::shared_ptr handler(new NoopEventHandler()); auto request_span = get_tracer()->StartSpan(__func__); trace_api::Scope scope(request_span); @@ -126,7 +126,7 @@ void send_request(curl::HttpClient &client, const std::string &url, const std::s request->AddHeader(hdr.first, hdr.second); } - session->SendRequest(*handler); + session->SendRequest(handler); session->FinishSession(); } From 14f0e7e290b1080f211a471b88496814d93aeed1 Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Mon, 14 Mar 2022 14:36:12 +0530 Subject: [PATCH 5/7] Incorporated Review Comments, Added wait on shutdown and force flush --- .../exporters/elasticsearch/es_log_exporter.h | 5 +- .../elasticsearch/src/es_log_exporter.cc | 2 + .../memory/in_memory_span_exporter.h | 5 +- .../exporters/zipkin/zipkin_exporter.h | 5 +- .../sdk/logs/batch_log_processor.h | 18 ++- sdk/include/opentelemetry/sdk/logs/exporter.h | 7 +- .../sdk/logs/simple_log_processor.h | 3 +- .../sdk/trace/batch_span_processor.h | 18 ++- .../sdk/trace/simple_processor.h | 20 +++- sdk/src/logs/batch_log_processor.cc | 60 ++++++++-- sdk/src/logs/simple_log_processor.cc | 17 ++- sdk/src/trace/batch_span_processor.cc | 52 ++++++-- sdk/test/logs/batch_log_processor_test.cc | 111 +++++++++++++++++- sdk/test/trace/batch_span_processor_test.cc | 95 ++++++++++++++- 14 files changed, 366 insertions(+), 52 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h index 14118b2f82..ad61281b7f 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h @@ -90,8 +90,9 @@ class ElasticsearchLogExporter final : public opentelemetry::sdk::logs::LogExpor &records) noexcept override; /** - * - * + * Exports a vector of log records to the Elasticsearch instance asynchronously. + * @param records A list of log records to send to Elasticsearch. + * @param result_callback callback function accepting ExportResult as argument */ void Export( const opentelemetry::nostd::span> &records, diff --git a/exporters/elasticsearch/src/es_log_exporter.cc b/exporters/elasticsearch/src/es_log_exporter.cc index 4c8969ae33..892405e002 100644 --- a/exporters/elasticsearch/src/es_log_exporter.cc +++ b/exporters/elasticsearch/src/es_log_exporter.cc @@ -175,6 +175,8 @@ class AsyncResponseHandler : public http_client::EventHandler case http_client::SessionState::NetworkError: OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Network error to elasticsearch"); break; + default: + break; } result_callback_(sdk::common::ExportResult::kFailure); } diff --git a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h index 565a118a51..73f090215a 100644 --- a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h +++ b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h @@ -65,8 +65,9 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte } /** - * - * + * Exports a batch of span recordables asynchronously. + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument */ void Export( const nostd::span> &spans, diff --git a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h index e350fb276a..810ac288e9 100644 --- a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h +++ b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h @@ -79,8 +79,9 @@ class ZipkinExporter final : public opentelemetry::sdk::trace::SpanExporter override; /** - * - * + * Export asynchronosly a batch of span recordables in JSON format + * @param spans a span of unique pointers to span recordables + * @param result_callback callback function accepting ExportResult as argument */ void Export( const nostd::span> &spans, diff --git a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h index fa462cb519..ba2df5d5a5 100644 --- a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h @@ -99,6 +99,19 @@ class BatchLogProcessor : public LogProcessor */ void DrainQueue(); + /** + * Should be called from Export to notify the main thread on Force Flush Completion + * @param was_force_flush_called - A flag to check if the current export is the result + * of a call to ForceFlush method. If true, then we have to + * notify the main thread to wake it up in the ForceFlush + * method. + */ + void NotifyForceFlushCompletion(const bool was_for_flush_called); + + /* In case of async export, wait and notify for shutdown to be completed.*/ + void WaitForShutdownCompletion(); + void NotifyShutdownCompletion(); + /* The configured backend log exporter */ std::unique_ptr exporter_; @@ -109,8 +122,8 @@ class BatchLogProcessor : public LogProcessor const bool is_export_async_; /* Synchronization primitives */ - std::condition_variable cv_, force_flush_cv_; - std::mutex cv_m_, force_flush_cv_m_, shutdown_m_; + std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_;; + std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_; /* The buffer/queue to which the ended logs are added */ common::CircularBuffer buffer_; @@ -119,6 +132,7 @@ class BatchLogProcessor : public LogProcessor std::atomic is_shutdown_{false}; std::atomic is_force_flush_{false}; std::atomic is_force_flush_notified_{false}; + std::atomic is_async_shutdown_notified_{false}; /* The background worker thread */ std::thread worker_thread_; diff --git a/sdk/include/opentelemetry/sdk/logs/exporter.h b/sdk/include/opentelemetry/sdk/logs/exporter.h index 829c9e1432..86f121d581 100644 --- a/sdk/include/opentelemetry/sdk/logs/exporter.h +++ b/sdk/include/opentelemetry/sdk/logs/exporter.h @@ -48,10 +48,9 @@ class LogExporter /** - * Exports the batch of log records to their export destination - * - * - * + * Exports asynchronously the batch of log records to their export destination + * @param records a span of unique pointers to log records + * @param result_callback callback function accepting ExportResult as argument */ virtual void Export( const nostd::span> &records, diff --git a/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h b/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h index cc3aec47b2..b377b86fc8 100644 --- a/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h @@ -28,7 +28,7 @@ class SimpleLogProcessor : public LogProcessor { public: - explicit SimpleLogProcessor(std::unique_ptr &&exporter); + explicit SimpleLogProcessor(std::unique_ptr &&exporter, bool is_export_async = false); virtual ~SimpleLogProcessor() = default; std::unique_ptr MakeRecordable() noexcept override; @@ -48,6 +48,7 @@ class SimpleLogProcessor : public LogProcessor opentelemetry::common::SpinLockMutex lock_; // The atomic boolean flag to ensure the ShutDown() function is only called once std::atomic_flag shutdown_latch_ = ATOMIC_FLAG_INIT; + bool is_export_async_ = false; }; } // namespace logs } // namespace sdk diff --git a/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h b/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h index b134e877ef..3ab4482b2e 100644 --- a/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h +++ b/sdk/include/opentelemetry/sdk/trace/batch_span_processor.h @@ -135,6 +135,19 @@ class BatchSpanProcessor : public SpanProcessor */ void DrainQueue(); + /** + * Should be called from Export to notify the main thread on Force Flush Completion + * @param was_force_flush_called - A flag to check if the current export is the result + * of a call to ForceFlush method. If true, then we have to + * notify the main thread to wake it up in the ForceFlush + * method. + */ + void NotifyForceFlushCompletion(const bool was_for_flush_called); + + /* In case of async export, wait and notify for shutdown to be completed.*/ + void WaitForShutdownCompletion(); + void NotifyShutdownCompletion(); + /* The configured backend exporter */ std::unique_ptr exporter_; @@ -145,8 +158,8 @@ class BatchSpanProcessor : public SpanProcessor const bool is_export_async_; /* Synchronization primitives */ - std::condition_variable cv_, force_flush_cv_; - std::mutex cv_m_, force_flush_cv_m_, shutdown_m_; + std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_; + std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_; /* The buffer/queue to which the ended spans are added */ common::CircularBuffer buffer_; @@ -155,6 +168,7 @@ class BatchSpanProcessor : public SpanProcessor std::atomic is_shutdown_{false}; std::atomic is_force_flush_{false}; std::atomic is_force_flush_notified_{false}; + std::atomic is_async_shutdown_notified_{false}; /* The background worker thread */ std::thread worker_thread_; diff --git a/sdk/include/opentelemetry/sdk/trace/simple_processor.h b/sdk/include/opentelemetry/sdk/trace/simple_processor.h index accc685965..df70b99e41 100644 --- a/sdk/include/opentelemetry/sdk/trace/simple_processor.h +++ b/sdk/include/opentelemetry/sdk/trace/simple_processor.h @@ -31,8 +31,9 @@ class SimpleSpanProcessor : public SpanProcessor * Initialize a simple span processor. * @param exporter the exporter used by the span processor */ - explicit SimpleSpanProcessor(std::unique_ptr &&exporter) noexcept + explicit SimpleSpanProcessor(std::unique_ptr &&exporter, bool is_export_async = false) noexcept : exporter_(std::move(exporter)) + , is_export_async_(is_export_async) {} std::unique_ptr MakeRecordable() noexcept override @@ -48,10 +49,18 @@ class SimpleSpanProcessor : public SpanProcessor { nostd::span> batch(&span, 1); const std::lock_guard locked(lock_); - if (exporter_->Export(batch) == sdk::common::ExportResult::kFailure) - { - /* Once it is defined how the SDK does logging, an error should be - * logged in this case. */ + if (is_export_async_ == false) { + if (exporter_->Export(batch) == sdk::common::ExportResult::kFailure) + { + /* Once it is defined how the SDK does logging, an error should be + * logged in this case. */ + } + } else { + exporter_->Export(batch, [](sdk::common::ExportResult result){ + /* Log the result + */ + return true; + }); } } @@ -78,6 +87,7 @@ class SimpleSpanProcessor : public SpanProcessor std::unique_ptr exporter_; opentelemetry::common::SpinLockMutex lock_; std::atomic_flag shutdown_latch_ = ATOMIC_FLAG_INIT; + bool is_export_async_ = false; }; } // namespace trace } // namespace sdk diff --git a/sdk/src/logs/batch_log_processor.cc b/sdk/src/logs/batch_log_processor.cc index 880e9bbb2f..a66368661b 100644 --- a/sdk/src/logs/batch_log_processor.cc +++ b/sdk/src/logs/batch_log_processor.cc @@ -153,35 +153,71 @@ void BatchLogProcessor::Export(const bool was_force_flush_called) return true; }); }); - if (is_export_async_ == false || was_force_flush_called == true) { + + if (is_export_async_ == false) { exporter_->Export( nostd::span>(records_arr.data(), records_arr.size())); - - // Notify the main thread in case this export was the result of a force flush. - if (was_force_flush_called == true) - { - is_force_flush_notified_ = true; - while (is_force_flush_notified_.load() == true) - { - force_flush_cv_.notify_one(); - } - } + NotifyForceFlushCompletion(was_force_flush_called); } else { exporter_->Export( nostd::span>(records_arr.data(), records_arr.size()), - [](sdk::common::ExportResult result) { + [this, was_force_flush_called](sdk::common::ExportResult result) { // TODO: Print result + NotifyForceFlushCompletion(was_force_flush_called); + + // Notify the thread which is waiting on shutdown to complete. + NotifyShutdownCompletion(); return true; }); } } +void BatchLogProcessor::NotifyForceFlushCompletion(const bool was_force_flush_called) +{ + // Notify the main thread in case this export was the result of a force flush. + if (was_force_flush_called == true) + { + is_force_flush_notified_ = true; + while (is_force_flush_notified_.load() == true) + { + force_flush_cv_.notify_one(); + } + } +} + +void BatchLogProcessor::WaitForShutdownCompletion() +{ + // Since async export is invoked due to shutdown, need to wait + // for async thread to complete. + if (is_export_async_) + { + std::unique_lock lk(async_shutdown_m_); + while (is_async_shutdown_notified_.load() == false) + { + async_shutdown_cv_.wait(lk); + } + } +} + +void BatchLogProcessor::NotifyShutdownCompletion() +{ + // Notify the thread which is waiting on shutdown to complete. + if (is_shutdown_.load() == true) { + is_async_shutdown_notified_.store(true); + async_shutdown_cv_.notify_one(); + } +} + void BatchLogProcessor::DrainQueue() { while (buffer_.empty() == false) { Export(false); + + // Since async export is invoked due to shutdown, need to wait + // for async thread to complete. + WaitForShutdownCompletion(); } } diff --git a/sdk/src/logs/simple_log_processor.cc b/sdk/src/logs/simple_log_processor.cc index 6e2fde9f14..dbafd29fc2 100644 --- a/sdk/src/logs/simple_log_processor.cc +++ b/sdk/src/logs/simple_log_processor.cc @@ -16,8 +16,9 @@ namespace logs * Initialize a simple log processor. * @param exporter the configured exporter where log records are sent */ -SimpleLogProcessor::SimpleLogProcessor(std::unique_ptr &&exporter) +SimpleLogProcessor::SimpleLogProcessor(std::unique_ptr &&exporter, bool is_export_async) : exporter_(std::move(exporter)) + , is_export_async_(is_export_async) {} std::unique_ptr SimpleLogProcessor::MakeRecordable() noexcept @@ -35,9 +36,17 @@ void SimpleLogProcessor::OnReceive(std::unique_ptr &&record) noexcep // Get lock to ensure Export() is never called concurrently const std::lock_guard locked(lock_); - if (exporter_->Export(batch) != sdk::common::ExportResult::kSuccess) - { - /* Alert user of the failed export */ + if (is_export_async_ == false) { + if (exporter_->Export(batch) != sdk::common::ExportResult::kSuccess) + { + /* Alert user of the failed export */ + } + } else { + exporter_->Export(batch, [](sdk::common::ExportResult result){ + /* Log the result + */ + return true; + }); } } /** diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index 5c9801a441..41546e994f 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -164,30 +164,62 @@ void BatchSpanProcessor::Export(const bool was_force_flush_called) if (is_export_async_ == false || was_force_flush_called == true) { exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size())); - // Notify the main thread in case this export was the result of a force flush. - if (was_force_flush_called == true) - { - is_force_flush_notified_ = true; - while (is_force_flush_notified_.load() == true) - { - force_flush_cv_.notify_one(); - } - } + NotifyForceFlushCompletion(was_force_flush_called); } else { exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size()), - [](sdk::common::ExportResult result) { + [this, was_force_flush_called](sdk::common::ExportResult result) { // TODO: Print result + NotifyForceFlushCompletion(was_force_flush_called); + // If export was called due to shutdown, notify the worker thread + NotifyShutdownCompletion(); return true; }); } } +void BatchSpanProcessor::NotifyForceFlushCompletion(const bool was_force_flush_called) +{ + // Notify the main thread in case this export was the result of a force flush. + if (was_force_flush_called == true) + { + is_force_flush_notified_ = true; + while (is_force_flush_notified_.load() == true) + { + force_flush_cv_.notify_one(); + } + } +} + +void BatchSpanProcessor::WaitForShutdownCompletion() +{ + // Since async export is invoked due to shutdown, need to wait + // for async thread to complete. + if (is_export_async_) + { + std::unique_lock lk(async_shutdown_m_); + while (is_async_shutdown_notified_.load() == false) + { + async_shutdown_cv_.wait(lk); + } + } +} + +void BatchSpanProcessor::NotifyShutdownCompletion() +{ + // Notify the thread which is waiting on shutdown to complete. + if (is_shutdown_.load() == true) { + is_async_shutdown_notified_.store(true); + async_shutdown_cv_.notify_one(); + } +} + void BatchSpanProcessor::DrainQueue() { while (buffer_.empty() == false) { Export(false); + WaitForShutdownCompletion(); } } diff --git a/sdk/test/logs/batch_log_processor_test.cc b/sdk/test/logs/batch_log_processor_test.cc index 2a8dc13bf3..c75ff72899 100644 --- a/sdk/test/logs/batch_log_processor_test.cc +++ b/sdk/test/logs/batch_log_processor_test.cc @@ -58,8 +58,11 @@ class MockLogExporter final : public LogExporter void Export(const opentelemetry::nostd::span> &records, opentelemetry::nostd::function_ref result_callback) noexcept override { - auto result = Export(records); - result_callback(result); + auto th = std::thread([this, records, result_callback](){ + auto result = Export(records); + result_callback(result); + }); + th.join(); } // toggles the boolean flag marking this exporter as shut down @@ -93,12 +96,14 @@ class BatchLogProcessorTest : public testing::Test // ::testing::Test const std::chrono::milliseconds export_delay = std::chrono::milliseconds(0), const std::chrono::milliseconds scheduled_delay_millis = std::chrono::milliseconds(5000), const size_t max_queue_size = 2048, - const size_t max_export_batch_size = 512) + const size_t max_export_batch_size = 512, + const bool is_export_async = false) { return std::shared_ptr( new BatchLogProcessor(std::unique_ptr(new MockLogExporter( logs_received, is_shutdown, is_export_completed, export_delay)), - max_queue_size, scheduled_delay_millis, max_export_batch_size)); + max_queue_size, scheduled_delay_millis, max_export_batch_size, + is_export_async)); } }; @@ -140,6 +145,53 @@ TEST_F(BatchLogProcessorTest, TestShutdown) EXPECT_TRUE(is_shutdown->load()); } +TEST_F(BatchLogProcessorTest, TestAsyncShutdown) +{ + // initialize a batch log processor with the test exporter + std::shared_ptr>> logs_received( + new std::vector>); + std::shared_ptr> is_shutdown(new std::atomic(false)); + std::shared_ptr> is_export_completed(new std::atomic(false)); + + const std::chrono::milliseconds export_delay(0); + const std::chrono::milliseconds scheduled_delay_millis(5000); + const size_t max_export_batch_size = 512; + const size_t max_queue_size = 2048; + const bool is_export_async = true; + + auto batch_processor = GetMockProcessor(logs_received, is_shutdown, is_export_completed, + export_delay, scheduled_delay_millis, max_queue_size, max_export_batch_size, + is_export_async); + + // Create a few test log records and send them to the processor + const int num_logs = 3; + + for (int i = 0; i < num_logs; ++i) + { + auto log = batch_processor->MakeRecordable(); + log->SetName("Log" + std::to_string(i)); + batch_processor->OnReceive(std::move(log)); + } + + // Test that shutting down the processor will first wait for the + // current batch of logs to be sent to the log exporter + // by checking the number of logs sent and the names of the logs sent + EXPECT_EQ(true, batch_processor->Shutdown()); + // It's safe to shutdown again + EXPECT_TRUE(batch_processor->Shutdown()); + + EXPECT_EQ(num_logs, logs_received->size()); + + // Assume logs are received by exporter in same order as sent by processor + for (int i = 0; i < num_logs; ++i) + { + EXPECT_EQ("Log" + std::to_string(i), logs_received->at(i)->GetName()); + } + + // Also check that the processor is shut down at the end + EXPECT_TRUE(is_shutdown->load()); +} + TEST_F(BatchLogProcessorTest, TestForceFlush) { std::shared_ptr> is_shutdown(new std::atomic(false)); @@ -181,6 +233,57 @@ TEST_F(BatchLogProcessorTest, TestForceFlush) } } +TEST_F(BatchLogProcessorTest, TestAsyncForceFlush) +{ + std::shared_ptr> is_shutdown(new std::atomic(false)); + std::shared_ptr>> logs_received( + new std::vector>); + std::shared_ptr> is_export_completed(new std::atomic(false)); + + const std::chrono::milliseconds export_delay(0); + const std::chrono::milliseconds scheduled_delay_millis(5000); + const size_t max_export_batch_size = 512; + const size_t max_queue_size = 2048; + const bool is_export_async = true; + + auto batch_processor = GetMockProcessor(logs_received, is_shutdown, is_export_completed, + export_delay, scheduled_delay_millis, max_queue_size, max_export_batch_size, + is_export_async); + + const int num_logs = 2048; + + for (int i = 0; i < num_logs; ++i) + { + auto log = batch_processor->MakeRecordable(); + log->SetName("Log" + std::to_string(i)); + batch_processor->OnReceive(std::move(log)); + } + + EXPECT_TRUE(batch_processor->ForceFlush()); + + EXPECT_EQ(num_logs, logs_received->size()); + for (int i = 0; i < num_logs; ++i) + { + EXPECT_EQ("Log" + std::to_string(i), logs_received->at(i)->GetName()); + } + + // Create some more logs to make sure that the processor still works + for (int i = 0; i < num_logs; ++i) + { + auto log = batch_processor->MakeRecordable(); + log->SetName("Log" + std::to_string(i)); + batch_processor->OnReceive(std::move(log)); + } + + EXPECT_TRUE(batch_processor->ForceFlush()); + + EXPECT_EQ(num_logs * 2, logs_received->size()); + for (int i = 0; i < num_logs * 2; ++i) + { + EXPECT_EQ("Log" + std::to_string(i % num_logs), logs_received->at(i)->GetName()); + } +} + TEST_F(BatchLogProcessorTest, TestManyLogsLoss) { /* Test that when exporting more than max_queue_size logs, some are most likely lost*/ diff --git a/sdk/test/trace/batch_span_processor_test.cc b/sdk/test/trace/batch_span_processor_test.cc index a445315aa2..fc6f5f4441 100644 --- a/sdk/test/trace/batch_span_processor_test.cc +++ b/sdk/test/trace/batch_span_processor_test.cc @@ -60,8 +60,11 @@ class MockSpanExporter final : public sdk::trace::SpanExporter const nostd::span> &spans, nostd::function_ref result_callback) noexcept override { - auto result = Export(spans); - result_callback(result); + auto th = std::thread([this, spans, result_callback](){ + auto result = Export(spans); + result_callback(result); + }); + th.join(); } bool Shutdown( @@ -139,7 +142,95 @@ TEST_F(BatchSpanProcessorTestPeer, TestShutdown) EXPECT_TRUE(is_shutdown->load()); } +TEST_F(BatchSpanProcessorTestPeer, TestAsyncShutdown) +{ + std::shared_ptr> is_shutdown(new std::atomic(false)); + std::shared_ptr>> spans_received( + new std::vector>); + + sdk::trace::BatchSpanProcessorOptions options{}; + options.is_export_async = true; + + auto batch_processor = + std::shared_ptr(new sdk::trace::BatchSpanProcessor( + std::unique_ptr(new MockSpanExporter(spans_received, is_shutdown)), + options)); + const int num_spans = 3; + + auto test_spans = GetTestSpans(batch_processor, num_spans); + + for (int i = 0; i < num_spans; ++i) + { + batch_processor->OnEnd(std::move(test_spans->at(i))); + } + + EXPECT_TRUE(batch_processor->Shutdown()); + // It's safe to shutdown again + EXPECT_TRUE(batch_processor->Shutdown()); + + EXPECT_EQ(num_spans, spans_received->size()); + for (int i = 0; i < num_spans; ++i) + { + EXPECT_EQ("Span " + std::to_string(i), spans_received->at(i)->GetName()); + } + + EXPECT_TRUE(is_shutdown->load()); +} + TEST_F(BatchSpanProcessorTestPeer, TestForceFlush) +{ + std::shared_ptr> is_shutdown(new std::atomic(false)); + std::shared_ptr>> spans_received( + new std::vector>); + + sdk::trace::BatchSpanProcessorOptions options{}; + options.is_export_async = true; + + auto batch_processor = + std::shared_ptr(new sdk::trace::BatchSpanProcessor( + std::unique_ptr(new MockSpanExporter(spans_received, is_shutdown)), + options)); + const int num_spans = 2048; + + auto test_spans = GetTestSpans(batch_processor, num_spans); + + for (int i = 0; i < num_spans; ++i) + { + batch_processor->OnEnd(std::move(test_spans->at(i))); + } + + // Give some time to export + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + EXPECT_TRUE(batch_processor->ForceFlush()); + + EXPECT_EQ(num_spans, spans_received->size()); + for (int i = 0; i < num_spans; ++i) + { + EXPECT_EQ("Span " + std::to_string(i), spans_received->at(i)->GetName()); + } + + // Create some more spans to make sure that the processor still works + auto more_test_spans = GetTestSpans(batch_processor, num_spans); + for (int i = 0; i < num_spans; ++i) + { + batch_processor->OnEnd(std::move(more_test_spans->at(i))); + } + + // Give some time to export the spans + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + EXPECT_TRUE(batch_processor->ForceFlush()); + + EXPECT_EQ(num_spans * 2, spans_received->size()); + for (int i = 0; i < num_spans; ++i) + { + EXPECT_EQ("Span " + std::to_string(i % num_spans), + spans_received->at(num_spans + i)->GetName()); + } +} + +TEST_F(BatchSpanProcessorTestPeer, TestAsyncForceFlush) { std::shared_ptr> is_shutdown(new std::atomic(false)); std::shared_ptr>> spans_received( From e6ea337718ea02f85545a3e495d2341615847f0a Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Mon, 14 Mar 2022 15:02:36 +0530 Subject: [PATCH 6/7] Fixing the format issue --- .../exporters/elasticsearch/es_log_exporter.h | 6 ++- .../elasticsearch/src/es_log_exporter.cc | 37 +++++++++---------- .../exporters/jaeger/jaeger_exporter.h | 8 ++-- exporters/jaeger/src/jaeger_exporter.cc | 6 +-- .../memory/in_memory_span_exporter.h | 3 +- .../exporters/ostream/log_exporter.h | 7 ++-- .../exporters/ostream/span_exporter.h | 7 ++-- exporters/ostream/src/log_exporter.cc | 6 +-- exporters/ostream/src/span_exporter.cc | 5 ++- exporters/otlp/src/otlp_grpc_exporter.cc | 9 ++--- exporters/otlp/src/otlp_grpc_log_exporter.cc | 7 ++-- exporters/otlp/src/otlp_http_exporter.cc | 5 +-- exporters/otlp/src/otlp_http_log_exporter.cc | 4 +- .../otlp/test/otlp_http_exporter_test.cc | 8 ++-- .../otlp/test/otlp_http_log_exporter_test.cc | 8 ++-- .../exporters/zipkin/zipkin_exporter.h | 7 ++-- exporters/zipkin/src/zipkin_exporter.cc | 5 +-- ext/test/http/curl_http_test.cc | 1 - .../sdk/logs/batch_log_processor.h | 3 +- sdk/include/opentelemetry/sdk/logs/exporter.h | 1 - .../sdk/logs/simple_log_processor.h | 5 ++- .../sdk/trace/simple_processor.h | 19 ++++++---- sdk/src/logs/batch_log_processor.cc | 9 +++-- sdk/src/logs/simple_log_processor.cc | 17 +++++---- sdk/src/trace/batch_span_processor.cc | 23 +++++++----- sdk/test/logs/batch_log_processor_test.cc | 34 ++++++++--------- sdk/test/logs/simple_log_processor_test.cc | 4 +- sdk/test/trace/batch_span_processor_test.cc | 2 +- sdk/test/trace/simple_processor_test.cc | 7 ++-- 29 files changed, 134 insertions(+), 129 deletions(-) diff --git a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h index ad61281b7f..50e763e0c2 100644 --- a/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h +++ b/exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h @@ -95,8 +95,10 @@ class ElasticsearchLogExporter final : public opentelemetry::sdk::logs::LogExpor * @param result_callback callback function accepting ExportResult as argument */ void Export( - const opentelemetry::nostd::span> &records, - nostd::function_ref result_callback) noexcept override; + const opentelemetry::nostd::span> + &records, + nostd::function_ref result_callback) noexcept + override; /** * Shutdown this exporter. diff --git a/exporters/elasticsearch/src/es_log_exporter.cc b/exporters/elasticsearch/src/es_log_exporter.cc index 892405e002..991678d6b9 100644 --- a/exporters/elasticsearch/src/es_log_exporter.cc +++ b/exporters/elasticsearch/src/es_log_exporter.cc @@ -110,7 +110,6 @@ class ResponseHandler : public http_client::EventHandler bool console_debug_ = false; }; - /** * This class handles the async response message from the Elasticsearch request */ @@ -121,20 +120,18 @@ class AsyncResponseHandler : public http_client::EventHandler * Creates a response handler, that by default doesn't display to console */ AsyncResponseHandler( - std::shared_ptr session, - nostd::function_ref result_callback, - bool console_debug = false) - : console_debug_{console_debug} - , session_{std::move(session)} - , result_callback_{result_callback} {} + std::shared_ptr session, + nostd::function_ref result_callback, + bool console_debug = false) + : console_debug_{console_debug}, + session_{std::move(session)}, + result_callback_{result_callback} + {} /** * Cleans up the session in the destructor. */ - ~AsyncResponseHandler() - { - session_->FinishSession(); - } + ~AsyncResponseHandler() { session_->FinishSession(); } /** * Automatically called when the response is received @@ -147,10 +144,12 @@ class AsyncResponseHandler : public http_client::EventHandler if (body_.find("\"failed\" : 0") == std::string::npos) { OTEL_INTERNAL_LOG_ERROR( - "[ES Trace Exporter] Logs were not written to Elasticsearch correctly, response body: " - << body_); + "[ES Trace Exporter] Logs were not written to Elasticsearch correctly, response body: " + << body_); result_callback_(sdk::common::ExportResult::kFailure); - } else { + } + else + { result_callback_(sdk::common::ExportResult::kSuccess); } } @@ -194,7 +193,6 @@ class AsyncResponseHandler : public http_client::EventHandler bool console_debug_ = false; }; - ElasticsearchLogExporter::ElasticsearchLogExporter() : options_{ElasticsearchExporterOptions()}, http_client_{new ext::http::client::curl::HttpClient()} @@ -284,8 +282,9 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export( } void ElasticsearchLogExporter::Export( - const opentelemetry::nostd::span> &records, - nostd::function_ref result_callback) noexcept + const opentelemetry::nostd::span> + &records, + nostd::function_ref result_callback) noexcept { // Return failure if this exporter has been shutdown if (isShutdown()) @@ -322,8 +321,8 @@ void ElasticsearchLogExporter::Export( request->SetBody(body_vec); // Send the request - auto handler = std::make_shared( - session, result_callback, options_.console_debug_); + auto handler = + std::make_shared(session, result_callback, options_.console_debug_); session->SendRequest(handler); } diff --git a/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h b/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h index c2e17b97ff..eb3b4bd621 100644 --- a/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h +++ b/exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h @@ -66,11 +66,9 @@ class JaegerExporter final : public opentelemetry::sdk::trace::SpanExporter * @param spans a span of unique pointers to span recordables * @param result_callback callback function accepting ExportResult as argument */ - void Export( - const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept override; - + void Export(const nostd::span> &spans, + nostd::function_ref + result_callback) noexcept override; /** * Shutdown the exporter. diff --git a/exporters/jaeger/src/jaeger_exporter.cc b/exporters/jaeger/src/jaeger_exporter.cc index 60d8a80db9..4a028773ca 100644 --- a/exporters/jaeger/src/jaeger_exporter.cc +++ b/exporters/jaeger/src/jaeger_exporter.cc @@ -71,14 +71,12 @@ sdk_common::ExportResult JaegerExporter::Export( } void JaegerExporter::Export( - const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept { OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); auto status = Export(spans); result_callback(status); - } void JaegerExporter::InitializeEndpoint() diff --git a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h index 73f090215a..3ebd3b8e89 100644 --- a/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h +++ b/exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h @@ -71,8 +71,7 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte */ void Export( const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept override + nostd::function_ref result_callback) noexcept override { OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); auto status = Export(spans); diff --git a/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h b/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h index 2a1f7b6aa2..017d967c70 100644 --- a/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h +++ b/exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h @@ -42,10 +42,9 @@ class OStreamLogExporter final : public opentelemetry::sdk::logs::LogExporter /** * Exports a span of logs sent from the processor asynchronously. */ - void Export( - const opentelemetry::nostd::span> &records, - opentelemetry::nostd::function_ref result_callback) - noexcept; + void Export(const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref + result_callback) noexcept; /** * Marks the OStream Log Exporter as shut down. diff --git a/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h b/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h index d23a92f77f..5af47280be 100644 --- a/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h +++ b/exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h @@ -39,9 +39,10 @@ class OStreamSpanExporter final : public opentelemetry::sdk::trace::SpanExporter &spans) noexcept override; void Export( - const opentelemetry::nostd::span> &spans, - opentelemetry::nostd::function_ref result_callback) - noexcept override; + const opentelemetry::nostd::span> + &spans, + opentelemetry::nostd::function_ref + result_callback) noexcept override; bool Shutdown( std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override; diff --git a/exporters/ostream/src/log_exporter.cc b/exporters/ostream/src/log_exporter.cc index cc97073123..39d5dab4a3 100644 --- a/exporters/ostream/src/log_exporter.cc +++ b/exporters/ostream/src/log_exporter.cc @@ -181,9 +181,9 @@ sdk::common::ExportResult OStreamLogExporter::Export( } void OStreamLogExporter::Export( - const opentelemetry::nostd::span> &records, - opentelemetry::nostd::function_ref result_callback) - noexcept + const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref + result_callback) noexcept { // Do not have async support auto result = Export(records); diff --git a/exporters/ostream/src/span_exporter.cc b/exporters/ostream/src/span_exporter.cc index 2a97f830c3..24fcf6007b 100644 --- a/exporters/ostream/src/span_exporter.cc +++ b/exporters/ostream/src/span_exporter.cc @@ -97,8 +97,9 @@ sdk::common::ExportResult OStreamSpanExporter::Export( } void OStreamSpanExporter::Export( - const opentelemetry::nostd::span> &spans, - opentelemetry::nostd::function_ref result_callback) noexcept + const opentelemetry::nostd::span> &spans, + opentelemetry::nostd::function_ref + result_callback) noexcept { auto result = Export(spans); result_callback(result); diff --git a/exporters/otlp/src/otlp_grpc_exporter.cc b/exporters/otlp/src/otlp_grpc_exporter.cc index b06e788388..f191580b7f 100644 --- a/exporters/otlp/src/otlp_grpc_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_exporter.cc @@ -144,14 +144,13 @@ sdk::common::ExportResult OtlpGrpcExporter::Export( } void OtlpGrpcExporter::Export( - const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept { - OTEL_INTERNAL_LOG_WARN("[OTLP TRACE GRPC Exporter] async not supported. Making sync interface call"); + OTEL_INTERNAL_LOG_WARN( + "[OTLP TRACE GRPC Exporter] async not supported. Making sync interface call"); auto status = Export(spans); result_callback(status); - } bool OtlpGrpcExporter::Shutdown(std::chrono::microseconds timeout) noexcept diff --git a/exporters/otlp/src/otlp_grpc_log_exporter.cc b/exporters/otlp/src/otlp_grpc_log_exporter.cc index 91e2a56e85..b56f8d9caa 100644 --- a/exporters/otlp/src/otlp_grpc_log_exporter.cc +++ b/exporters/otlp/src/otlp_grpc_log_exporter.cc @@ -162,10 +162,11 @@ opentelemetry::sdk::common::ExportResult OtlpGrpcLogExporter::Export( } void OtlpGrpcLogExporter::Export( - const nostd::span> &logs, - nostd::function_ref result_callback) noexcept + const nostd::span> &logs, + nostd::function_ref result_callback) noexcept { - OTEL_INTERNAL_LOG_WARN("[OTLP LOG GRPC Exporter] async not supported. Making sync interface call"); + OTEL_INTERNAL_LOG_WARN( + "[OTLP LOG GRPC Exporter] async not supported. Making sync interface call"); auto status = Export(logs); result_callback(status); } diff --git a/exporters/otlp/src/otlp_http_exporter.cc b/exporters/otlp/src/otlp_http_exporter.cc index 452fa89094..1262533283 100644 --- a/exporters/otlp/src/otlp_http_exporter.cc +++ b/exporters/otlp/src/otlp_http_exporter.cc @@ -13,7 +13,6 @@ #include "opentelemetry/sdk/common/global_log_handler.h" - namespace nostd = opentelemetry::nostd; OPENTELEMETRY_BEGIN_NAMESPACE @@ -60,8 +59,8 @@ opentelemetry::sdk::common::ExportResult OtlpHttpExporter::Export( } void OtlpHttpExporter::Export( - const nostd::span> &spans, - nostd::function_ref result_callback) noexcept + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept { OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); auto status = Export(spans); diff --git a/exporters/otlp/src/otlp_http_log_exporter.cc b/exporters/otlp/src/otlp_http_log_exporter.cc index 734d6ecc48..bda33b4e31 100644 --- a/exporters/otlp/src/otlp_http_log_exporter.cc +++ b/exporters/otlp/src/otlp_http_log_exporter.cc @@ -60,8 +60,8 @@ opentelemetry::sdk::common::ExportResult OtlpHttpLogExporter::Export( } void OtlpHttpLogExporter::Export( - const nostd::span> &logs, - nostd::function_ref result_callback) noexcept + const nostd::span> &logs, + nostd::function_ref result_callback) noexcept { OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call"); auto status = Export(logs); diff --git a/exporters/otlp/test/otlp_http_exporter_test.cc b/exporters/otlp/test/otlp_http_exporter_test.cc index ef9f824733..04d37ea0b9 100644 --- a/exporters/otlp/test/otlp_http_exporter_test.cc +++ b/exporters/otlp/test/otlp_http_exporter_test.cc @@ -138,8 +138,8 @@ TEST_F(OtlpHttpExporterTestPeer, ExportJsonIntegrationTest) auto mock_session = std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) - .WillOnce([&mock_session, - report_trace_id](std::shared_ptr callback) { + .WillOnce([&mock_session, report_trace_id]( + std::shared_ptr callback) { auto check_json = nlohmann::json::parse(mock_session->GetRequest()->body_, nullptr, false); auto resource_span = *check_json["resource_spans"].begin(); auto instrumentation_library_span = *resource_span["instrumentation_library_spans"].begin(); @@ -217,8 +217,8 @@ TEST_F(OtlpHttpExporterTestPeer, ExportBinaryIntegrationTest) auto mock_session = std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) - .WillOnce([&mock_session, - report_trace_id](std::shared_ptr callback) { + .WillOnce([&mock_session, report_trace_id]( + std::shared_ptr callback) { opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest request_body; request_body.ParseFromArray(&mock_session->GetRequest()->body_[0], static_cast(mock_session->GetRequest()->body_.size())); diff --git a/exporters/otlp/test/otlp_http_log_exporter_test.cc b/exporters/otlp/test/otlp_http_log_exporter_test.cc index c7f52db214..07d7e33f33 100644 --- a/exporters/otlp/test/otlp_http_log_exporter_test.cc +++ b/exporters/otlp/test/otlp_http_log_exporter_test.cc @@ -148,8 +148,8 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportJsonIntegrationTest) auto mock_session = std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) - .WillOnce([&mock_session, report_trace_id, - report_span_id](std::shared_ptr callback) { + .WillOnce([&mock_session, report_trace_id, report_span_id]( + std::shared_ptr callback) { auto check_json = nlohmann::json::parse(mock_session->GetRequest()->body_, nullptr, false); auto resource_logs = *check_json["resource_logs"].begin(); auto instrumentation_library_span = *resource_logs["instrumentation_library_logs"].begin(); @@ -232,8 +232,8 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportBinaryIntegrationTest) auto mock_session = std::static_pointer_cast(no_send_client->session_); EXPECT_CALL(*mock_session, SendRequest) - .WillOnce([&mock_session, report_trace_id, - report_span_id](std::shared_ptr callback) { + .WillOnce([&mock_session, report_trace_id, report_span_id]( + std::shared_ptr callback) { opentelemetry::proto::collector::logs::v1::ExportLogsServiceRequest request_body; request_body.ParseFromArray(&mock_session->GetRequest()->body_[0], static_cast(mock_session->GetRequest()->body_.size())); diff --git a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h index 810ac288e9..3ac8c04028 100644 --- a/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h +++ b/exporters/zipkin/include/opentelemetry/exporters/zipkin/zipkin_exporter.h @@ -83,10 +83,9 @@ class ZipkinExporter final : public opentelemetry::sdk::trace::SpanExporter * @param spans a span of unique pointers to span recordables * @param result_callback callback function accepting ExportResult as argument */ - void Export( - const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept override; + void Export(const nostd::span> &spans, + nostd::function_ref + result_callback) noexcept override; /** * Shut down the exporter. diff --git a/exporters/zipkin/src/zipkin_exporter.cc b/exporters/zipkin/src/zipkin_exporter.cc index e3341945c8..050ed4c9a8 100644 --- a/exporters/zipkin/src/zipkin_exporter.cc +++ b/exporters/zipkin/src/zipkin_exporter.cc @@ -94,9 +94,8 @@ sdk::common::ExportResult ZipkinExporter::Export( } void ZipkinExporter::Export( - const nostd::span> &spans, - nostd::function_ref result_callback) - noexcept + const nostd::span> &spans, + nostd::function_ref result_callback) noexcept { OTEL_INTERNAL_LOG_WARN("[ZIPKIN EXPORTER] async not supported. Making sync interface call"); auto status = Export(spans); diff --git a/ext/test/http/curl_http_test.cc b/ext/test/http/curl_http_test.cc index 3b48d2aa1e..7085a1da33 100644 --- a/ext/test/http/curl_http_test.cc +++ b/ext/test/http/curl_http_test.cc @@ -226,7 +226,6 @@ TEST_F(BasicCurlHttpTests, SendPostRequest) session_manager->CancelAllSessions(); session_manager->FinishAllSessions(); - } TEST_F(BasicCurlHttpTests, RequestTimeout) diff --git a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h index ba2df5d5a5..569f530430 100644 --- a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h @@ -122,7 +122,8 @@ class BatchLogProcessor : public LogProcessor const bool is_export_async_; /* Synchronization primitives */ - std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_;; + std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_; + ; std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_; /* The buffer/queue to which the ended logs are added */ diff --git a/sdk/include/opentelemetry/sdk/logs/exporter.h b/sdk/include/opentelemetry/sdk/logs/exporter.h index 86f121d581..05990471d0 100644 --- a/sdk/include/opentelemetry/sdk/logs/exporter.h +++ b/sdk/include/opentelemetry/sdk/logs/exporter.h @@ -46,7 +46,6 @@ class LogExporter virtual sdk::common::ExportResult Export( const nostd::span> &records) noexcept = 0; - /** * Exports asynchronously the batch of log records to their export destination * @param records a span of unique pointers to log records diff --git a/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h b/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h index b377b86fc8..28fcca78a6 100644 --- a/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/simple_log_processor.h @@ -28,7 +28,8 @@ class SimpleLogProcessor : public LogProcessor { public: - explicit SimpleLogProcessor(std::unique_ptr &&exporter, bool is_export_async = false); + explicit SimpleLogProcessor(std::unique_ptr &&exporter, + bool is_export_async = false); virtual ~SimpleLogProcessor() = default; std::unique_ptr MakeRecordable() noexcept override; @@ -48,7 +49,7 @@ class SimpleLogProcessor : public LogProcessor opentelemetry::common::SpinLockMutex lock_; // The atomic boolean flag to ensure the ShutDown() function is only called once std::atomic_flag shutdown_latch_ = ATOMIC_FLAG_INIT; - bool is_export_async_ = false; + bool is_export_async_ = false; }; } // namespace logs } // namespace sdk diff --git a/sdk/include/opentelemetry/sdk/trace/simple_processor.h b/sdk/include/opentelemetry/sdk/trace/simple_processor.h index df70b99e41..982a432e0c 100644 --- a/sdk/include/opentelemetry/sdk/trace/simple_processor.h +++ b/sdk/include/opentelemetry/sdk/trace/simple_processor.h @@ -31,9 +31,9 @@ class SimpleSpanProcessor : public SpanProcessor * Initialize a simple span processor. * @param exporter the exporter used by the span processor */ - explicit SimpleSpanProcessor(std::unique_ptr &&exporter, bool is_export_async = false) noexcept - : exporter_(std::move(exporter)) - , is_export_async_(is_export_async) + explicit SimpleSpanProcessor(std::unique_ptr &&exporter, + bool is_export_async = false) noexcept + : exporter_(std::move(exporter)), is_export_async_(is_export_async) {} std::unique_ptr MakeRecordable() noexcept override @@ -49,16 +49,19 @@ class SimpleSpanProcessor : public SpanProcessor { nostd::span> batch(&span, 1); const std::lock_guard locked(lock_); - if (is_export_async_ == false) { + if (is_export_async_ == false) + { if (exporter_->Export(batch) == sdk::common::ExportResult::kFailure) { /* Once it is defined how the SDK does logging, an error should be * logged in this case. */ } - } else { - exporter_->Export(batch, [](sdk::common::ExportResult result){ + } + else + { + exporter_->Export(batch, [](sdk::common::ExportResult result) { /* Log the result - */ + */ return true; }); } @@ -87,7 +90,7 @@ class SimpleSpanProcessor : public SpanProcessor std::unique_ptr exporter_; opentelemetry::common::SpinLockMutex lock_; std::atomic_flag shutdown_latch_ = ATOMIC_FLAG_INIT; - bool is_export_async_ = false; + bool is_export_async_ = false; }; } // namespace trace } // namespace sdk diff --git a/sdk/src/logs/batch_log_processor.cc b/sdk/src/logs/batch_log_processor.cc index a66368661b..7d8782f115 100644 --- a/sdk/src/logs/batch_log_processor.cc +++ b/sdk/src/logs/batch_log_processor.cc @@ -154,12 +154,14 @@ void BatchLogProcessor::Export(const bool was_force_flush_called) }); }); - if (is_export_async_ == false) { + if (is_export_async_ == false) + { exporter_->Export( nostd::span>(records_arr.data(), records_arr.size())); NotifyForceFlushCompletion(was_force_flush_called); } - else { + else + { exporter_->Export( nostd::span>(records_arr.data(), records_arr.size()), [this, was_force_flush_called](sdk::common::ExportResult result) { @@ -203,7 +205,8 @@ void BatchLogProcessor::WaitForShutdownCompletion() void BatchLogProcessor::NotifyShutdownCompletion() { // Notify the thread which is waiting on shutdown to complete. - if (is_shutdown_.load() == true) { + if (is_shutdown_.load() == true) + { is_async_shutdown_notified_.store(true); async_shutdown_cv_.notify_one(); } diff --git a/sdk/src/logs/simple_log_processor.cc b/sdk/src/logs/simple_log_processor.cc index dbafd29fc2..e16a5e631e 100644 --- a/sdk/src/logs/simple_log_processor.cc +++ b/sdk/src/logs/simple_log_processor.cc @@ -16,9 +16,9 @@ namespace logs * Initialize a simple log processor. * @param exporter the configured exporter where log records are sent */ -SimpleLogProcessor::SimpleLogProcessor(std::unique_ptr &&exporter, bool is_export_async) - : exporter_(std::move(exporter)) - , is_export_async_(is_export_async) +SimpleLogProcessor::SimpleLogProcessor(std::unique_ptr &&exporter, + bool is_export_async) + : exporter_(std::move(exporter)), is_export_async_(is_export_async) {} std::unique_ptr SimpleLogProcessor::MakeRecordable() noexcept @@ -36,15 +36,18 @@ void SimpleLogProcessor::OnReceive(std::unique_ptr &&record) noexcep // Get lock to ensure Export() is never called concurrently const std::lock_guard locked(lock_); - if (is_export_async_ == false) { + if (is_export_async_ == false) + { if (exporter_->Export(batch) != sdk::common::ExportResult::kSuccess) { /* Alert user of the failed export */ } - } else { - exporter_->Export(batch, [](sdk::common::ExportResult result){ + } + else + { + exporter_->Export(batch, [](sdk::common::ExportResult result) { /* Log the result - */ + */ return true; }); } diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index 41546e994f..a5a54d6218 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -161,20 +161,22 @@ void BatchSpanProcessor::Export(const bool was_force_flush_called) /* Call the sync Export when force flush was called, even if is_export_async_ is true. */ - if (is_export_async_ == false || was_force_flush_called == true) { + if (is_export_async_ == false || was_force_flush_called == true) + { exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size())); NotifyForceFlushCompletion(was_force_flush_called); } - else { + else + { exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size()), - [this, was_force_flush_called](sdk::common::ExportResult result) { - // TODO: Print result - NotifyForceFlushCompletion(was_force_flush_called); - // If export was called due to shutdown, notify the worker thread - NotifyShutdownCompletion(); - return true; - }); + [this, was_force_flush_called](sdk::common::ExportResult result) { + // TODO: Print result + NotifyForceFlushCompletion(was_force_flush_called); + // If export was called due to shutdown, notify the worker thread + NotifyShutdownCompletion(); + return true; + }); } } @@ -208,7 +210,8 @@ void BatchSpanProcessor::WaitForShutdownCompletion() void BatchSpanProcessor::NotifyShutdownCompletion() { // Notify the thread which is waiting on shutdown to complete. - if (is_shutdown_.load() == true) { + if (is_shutdown_.load() == true) + { is_async_shutdown_notified_.store(true); async_shutdown_cv_.notify_one(); } diff --git a/sdk/test/logs/batch_log_processor_test.cc b/sdk/test/logs/batch_log_processor_test.cc index c75ff72899..55850bf414 100644 --- a/sdk/test/logs/batch_log_processor_test.cc +++ b/sdk/test/logs/batch_log_processor_test.cc @@ -55,10 +55,11 @@ class MockLogExporter final : public LogExporter return ExportResult::kSuccess; } - void Export(const opentelemetry::nostd::span> &records, - opentelemetry::nostd::function_ref result_callback) noexcept override + void Export( + const opentelemetry::nostd::span> &records, + opentelemetry::nostd::function_ref result_callback) noexcept override { - auto th = std::thread([this, records, result_callback](){ + auto th = std::thread([this, records, result_callback]() { auto result = Export(records); result_callback(result); }); @@ -99,11 +100,10 @@ class BatchLogProcessorTest : public testing::Test // ::testing::Test const size_t max_export_batch_size = 512, const bool is_export_async = false) { - return std::shared_ptr( - new BatchLogProcessor(std::unique_ptr(new MockLogExporter( - logs_received, is_shutdown, is_export_completed, export_delay)), - max_queue_size, scheduled_delay_millis, max_export_batch_size, - is_export_async)); + return std::shared_ptr(new BatchLogProcessor( + std::unique_ptr( + new MockLogExporter(logs_received, is_shutdown, is_export_completed, export_delay)), + max_queue_size, scheduled_delay_millis, max_export_batch_size, is_export_async)); } }; @@ -156,12 +156,12 @@ TEST_F(BatchLogProcessorTest, TestAsyncShutdown) const std::chrono::milliseconds export_delay(0); const std::chrono::milliseconds scheduled_delay_millis(5000); const size_t max_export_batch_size = 512; - const size_t max_queue_size = 2048; - const bool is_export_async = true; + const size_t max_queue_size = 2048; + const bool is_export_async = true; auto batch_processor = GetMockProcessor(logs_received, is_shutdown, is_export_completed, - export_delay, scheduled_delay_millis, max_queue_size, max_export_batch_size, - is_export_async); + export_delay, scheduled_delay_millis, max_queue_size, + max_export_batch_size, is_export_async); // Create a few test log records and send them to the processor const int num_logs = 3; @@ -243,14 +243,14 @@ TEST_F(BatchLogProcessorTest, TestAsyncForceFlush) const std::chrono::milliseconds export_delay(0); const std::chrono::milliseconds scheduled_delay_millis(5000); const size_t max_export_batch_size = 512; - const size_t max_queue_size = 2048; - const bool is_export_async = true; + const size_t max_queue_size = 2048; + const bool is_export_async = true; auto batch_processor = GetMockProcessor(logs_received, is_shutdown, is_export_completed, - export_delay, scheduled_delay_millis, max_queue_size, max_export_batch_size, - is_export_async); + export_delay, scheduled_delay_millis, max_queue_size, + max_export_batch_size, is_export_async); - const int num_logs = 2048; + const int num_logs = 2048; for (int i = 0; i < num_logs; ++i) { diff --git a/sdk/test/logs/simple_log_processor_test.cc b/sdk/test/logs/simple_log_processor_test.cc index 2a86910c94..8b8efa6636 100644 --- a/sdk/test/logs/simple_log_processor_test.cc +++ b/sdk/test/logs/simple_log_processor_test.cc @@ -55,7 +55,7 @@ class TestExporter final : public LogExporter // Dummy Async Export implementation void Export(const nostd::span> &records, - nostd::function_ref result_callback) noexcept override + nostd::function_ref result_callback) noexcept override { auto result = Export(records); result_callback(result); @@ -146,7 +146,7 @@ class FailShutDownExporter final : public LogExporter } void Export(const nostd::span> &records, - nostd::function_ref result_callback) noexcept override + nostd::function_ref result_callback) noexcept override { result_callback(ExportResult::kSuccess); } diff --git a/sdk/test/trace/batch_span_processor_test.cc b/sdk/test/trace/batch_span_processor_test.cc index fc6f5f4441..91561ac658 100644 --- a/sdk/test/trace/batch_span_processor_test.cc +++ b/sdk/test/trace/batch_span_processor_test.cc @@ -60,7 +60,7 @@ class MockSpanExporter final : public sdk::trace::SpanExporter const nostd::span> &spans, nostd::function_ref result_callback) noexcept override { - auto th = std::thread([this, spans, result_callback](){ + auto th = std::thread([this, spans, result_callback]() { auto result = Export(spans); result_callback(result); }); diff --git a/sdk/test/trace/simple_processor_test.cc b/sdk/test/trace/simple_processor_test.cc index b46339ebad..b8bad5962d 100644 --- a/sdk/test/trace/simple_processor_test.cc +++ b/sdk/test/trace/simple_processor_test.cc @@ -51,10 +51,9 @@ class RecordShutdownExporter final : public SpanExporter return ExportResult::kSuccess; } - void Export( - const opentelemetry::nostd::span> &spans, - opentelemetry::nostd::function_ref result_callback) - noexcept override + void Export(const opentelemetry::nostd::span> &spans, + opentelemetry::nostd::function_ref + result_callback) noexcept override { result_callback(ExportResult::kSuccess); } From e99263aab33d14a2b91f48c8ead29b5c77a068aa Mon Sep 17 00:00:00 2001 From: DebajitDas Date: Mon, 14 Mar 2022 16:40:29 +0530 Subject: [PATCH 7/7] Update minor nit --- sdk/include/opentelemetry/sdk/logs/batch_log_processor.h | 1 - sdk/src/trace/batch_span_processor.cc | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h index 569f530430..ca920a87d6 100644 --- a/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h +++ b/sdk/include/opentelemetry/sdk/logs/batch_log_processor.h @@ -123,7 +123,6 @@ class BatchLogProcessor : public LogProcessor /* Synchronization primitives */ std::condition_variable cv_, force_flush_cv_, async_shutdown_cv_; - ; std::mutex cv_m_, force_flush_cv_m_, shutdown_m_, async_shutdown_m_; /* The buffer/queue to which the ended logs are added */ diff --git a/sdk/src/trace/batch_span_processor.cc b/sdk/src/trace/batch_span_processor.cc index a5a54d6218..05ce598bb8 100644 --- a/sdk/src/trace/batch_span_processor.cc +++ b/sdk/src/trace/batch_span_processor.cc @@ -161,7 +161,7 @@ void BatchSpanProcessor::Export(const bool was_force_flush_called) /* Call the sync Export when force flush was called, even if is_export_async_ is true. */ - if (is_export_async_ == false || was_force_flush_called == true) + if (is_export_async_ == false) { exporter_->Export(nostd::span>(spans_arr.data(), spans_arr.size()));