diff --git a/CHANGELOG.md b/CHANGELOG.md index c7268973db..b1c42c903a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,13 @@ Increment the: ## [Unreleased] +* [EXPORTER] Fix the Elasticsearch log exporter aborting the process when a log + record's body or attributes contain bytes that are not valid UTF-8. The + exporter now substitutes the replacement character for the invalid bytes + and continues, instead of `nlohmann::json::dump()` throwing out of a + `noexcept` function. + [#4439](https://github.com/open-telemetry/opentelemetry-cpp/issues/4439) + * [CONFIGURATION] Apply general `attribute_limits` per individual limit field. If a model-specific limit is set it is used, otherwise the matching general limit, otherwise the model-specific default. Limit fields on diff --git a/exporters/elasticsearch/src/es_log_record_exporter.cc b/exporters/elasticsearch/src/es_log_record_exporter.cc index af819c8eb7..cd3b1bdbfd 100644 --- a/exporters/elasticsearch/src/es_log_record_exporter.cc +++ b/exporters/elasticsearch/src/es_log_record_exporter.cc @@ -429,7 +429,13 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export( // Add the context of the Recordable auto json_record = std::unique_ptr( static_cast(record.release())); - body += json_record->GetJSON().dump() + "\n"; + // A log record's body or attributes may carry bytes that are not valid UTF-8 (e.g. a + // truncated multibyte sequence, or a payload read in another encoding). dump() throws + // on those by default, and Export() is noexcept, so the exception would otherwise + // terminate the process. error_handler_t::replace substitutes U+FFFD for the invalid + // bytes instead, so the record is still exported with everything else intact. + body += json_record->GetJSON().dump(-1, ' ', false, nlohmann::json::error_handler_t::replace) + + "\n"; } std::vector body_vec(body.begin(), body.end()); request->SetBody(body_vec); diff --git a/exporters/elasticsearch/test/es_log_record_exporter_test.cc b/exporters/elasticsearch/test/es_log_record_exporter_test.cc index a65c0b4c1c..52a5d9b6ef 100644 --- a/exporters/elasticsearch/test/es_log_record_exporter_test.cc +++ b/exporters/elasticsearch/test/es_log_record_exporter_test.cc @@ -4,7 +4,9 @@ #include "opentelemetry/exporters/elasticsearch/es_log_record_exporter.h" #include "opentelemetry/common/timestamp.h" #include "opentelemetry/exporters/elasticsearch/es_log_recordable.h" +#include "opentelemetry/ext/http/client/http_client.h" #include "opentelemetry/logs/severity.h" +#include "opentelemetry/nostd/function_ref.h" #include "opentelemetry/nostd/span.h" #include "opentelemetry/nostd/string_view.h" #include "opentelemetry/nostd/utility.h" @@ -23,11 +25,130 @@ #include #include "nlohmann/json.hpp" +namespace +{ +namespace http_client = opentelemetry::ext::http::client; + +// A response shaped like a successful Elasticsearch bulk reply: the exporter looks for +// `"failed" : 0` in the body (see ElasticsearchLogRecordExporter::Export) in addition to the +// status code before reporting success. +class FakeResponse final : public http_client::Response +{ +public: + FakeResponse() + { + static const std::string kSuccessBody = R"({"errors": false, "failed" : 0})"; + body_.assign(kSuccessBody.begin(), kSuccessBody.end()); + } + + const http_client::Body &GetBody() const noexcept override { return body_; } + + bool ForEachHeader(opentelemetry::nostd::function_ref) + const noexcept override + { + return true; + } + + bool ForEachHeader(const opentelemetry::nostd::string_view &, + opentelemetry::nostd::function_ref) + const noexcept override + { + return true; + } + + http_client::StatusCode GetStatusCode() const noexcept override { return 200; } + +private: + http_client::Body body_; +}; + +// A request that accepts and discards everything set on it: the synchronous export path only +// needs a Request to exist, not to inspect what was written to it. +class FakeRequest final : public http_client::Request +{ +public: + void SetMethod(http_client::Method) noexcept override {} + void SetUri(opentelemetry::nostd::string_view) noexcept override {} + void SetSslOptions(const http_client::HttpSslOptions &) noexcept override {} + void SetBody(http_client::Body &) noexcept override {} + void AddHeader(opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) noexcept override + {} + void ReplaceHeader(opentelemetry::nostd::string_view, + opentelemetry::nostd::string_view) noexcept override + {} + void SetTimeoutMs(std::chrono::milliseconds) noexcept override {} + void SetCompression(const http_client::Compression &) noexcept override {} + void EnableLogging(bool) noexcept override {} + void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {} +}; + +// A session whose SendRequest() answers synchronously with a successful FakeResponse, so the +// exporter's own wait for a response returns immediately without needing a real connection. +class FakeSession final : public http_client::Session +{ +public: + std::shared_ptr CreateRequest() noexcept override + { + return std::make_shared(); + } + + void SendRequest(std::shared_ptr handler) noexcept override + { + FakeResponse response; + handler->OnResponse(response); + } + + bool IsSessionActive() noexcept override { return true; } + bool CancelSession() noexcept override { return true; } + bool FinishSession() noexcept override { return true; } +}; + +class FakeHttpClient final : public http_client::HttpClient +{ +public: + std::shared_ptr CreateSession( + opentelemetry::nostd::string_view) noexcept override + { + return std::make_shared(); + } + + bool CancelAllSessions() noexcept override { return true; } + bool FinishAllSessions() noexcept override { return true; } + void SetMaxSessionsPerConnection(std::size_t) noexcept override {} +}; + +} // namespace + namespace sdklogs = opentelemetry::sdk::logs; namespace logs_api = opentelemetry::logs; namespace nostd = opentelemetry::nostd; namespace logs_exporter = opentelemetry::exporter::logs; +// Regression test: a log record whose body carries bytes that are not valid UTF-8 used to +// abort the process. ElasticSearchRecordable::WriteValue stores the value as given, and +// Export() previously called nlohmann::json::dump() with its default strict error handler, +// which throws on invalid UTF-8; since Export() is noexcept, that throw became +// std::terminate(). The exporter now tolerates it instead of crashing. +TEST(ElasticsearchLogsExporterTests, ExportingARecordWithInvalidUtf8DoesNotAbort) +{ + logs_exporter::ElasticsearchExporterOptions options; + auto http_client = std::make_shared(); + auto exporter = std::unique_ptr( + new logs_exporter::ElasticsearchLogRecordExporter(options, http_client)); + + auto record = exporter->MakeRecordable(); + std::string body = "payload "; + body += "\xC3\x28"; // a two byte sequence that is not valid UTF-8 + record->SetBody(body); + + auto result = exporter->Export(nostd::span>(&record, 1)); + + EXPECT_EQ(result, opentelemetry::sdk::common::ExportResult::kSuccess); +} + TEST(ElasticsearchLogsExporterTests, CustomClientConstructionSucceeds) { logs_exporter::ElasticsearchExporterOptions opts;