async_client: API changes to reduce null pointer derefs (#13339)

As a follow-up to #13328, always creating a body buffer to avoid the pattern of null pointer derefs.

Risk Level: Medium
Testing: unit tests pass
Docs Changes: n/a
Release Notes: yes

Signed-off-by: Alyssa Wilk <alyssar@chromium.org>

docs/root/version_history/current.rst

@@ -17,6 +17,7 @@ Minor Behavior Changes
 *Changes that may cause incompatibilities for some users, but should not for most*
 
 * adaptive concurrency: added a response body / grpc-message header for rejected requests.
+* async_client: minor change to handling header only responses more similar to header-with-empty-body responses.
 * build: an :ref:`Ubuntu based debug image <install_binaries>` is built and published in DockerHub.
 * build: the debug information will be generated separately to reduce target size and reduce compilation time when build in compilation mode `dbg` and `opt`. Users will need to build dwp file to debug with gdb.
 * compressor: always insert `Vary` headers for compressible resources even if it's decided not to compress a response due to incompatible `Accept-Encoding` value. The `Vary` header needs to be inserted to let a caching proxy in front of Envoy know that the requested resource still can be served with compression applied.

include/envoy/http/message.h

@@ -22,10 +22,9 @@ template <class HeaderType, class TrailerType> class Message {
   virtual HeaderType& headers() PURE;
 
   /**
-   * @return Buffer::InstancePtr& the message body, if any. Callers are free to reallocate, remove,
-   *         etc. the body.
+   * @return Buffer::Instance the message body, if any. Callers are free to modify the body.
    */
-  virtual Buffer::InstancePtr& body() PURE;
+  virtual Buffer::Instance& body() PURE;
 
   /**
    * @return TrailerType* the message trailers, if any.

source/common/config/http_subscription_impl.cc

@@ -70,10 +70,9 @@ void HttpSubscriptionImpl::createRequest(Http::RequestMessage& request) {
   stats_.update_attempt_.inc();
   request.headers().setReferenceMethod(Http::Headers::get().MethodValues.Post);
   request.headers().setPath(path_);
-  request.body() = std::make_unique<Buffer::OwnedImpl>(
-      VersionConverter::getJsonStringFromMessage(request_, transport_api_version_));
+  request.body().add(VersionConverter::getJsonStringFromMessage(request_, transport_api_version_));
   request.headers().setReferenceContentType(Http::Headers::get().ContentTypeValues.Json);
-  request.headers().setContentLength(request.body()->length());
+  request.headers().setContentLength(request.body().length());
 }
 
 void HttpSubscriptionImpl::parseResponse(const Http::ResponseMessage& response) {

source/common/config/remote_data_fetcher.cc

@@ -44,15 +44,15 @@ void RemoteDataFetcher::onSuccess(const Http::AsyncClient::Request&,
   const uint64_t status_code = Http::Utility::getResponseStatus(response->headers());
   if (status_code == enumToInt(Http::Code::OK)) {
     ENVOY_LOG(debug, "fetch remote data [uri = {}]: success", uri_.uri());
-    if (response->body()) {
+    if (response->body().length() > 0) {
       auto& crypto_util = Envoy::Common::Crypto::UtilitySingleton::get();
-      const auto content_hash = Hex::encode(crypto_util.getSha256Digest(*response->body()));
+      const auto content_hash = Hex::encode(crypto_util.getSha256Digest(response->body()));
 
       if (content_hash_ != content_hash) {
         ENVOY_LOG(debug, "fetch remote data [uri = {}]: data is invalid", uri_.uri());
         callback_.onFailure(FailureReason::InvalidData);
       } else {
-        callback_.onSuccess(response->body()->toString());
+        callback_.onSuccess(response->bodyAsString());
       }
     } else {
       ENVOY_LOG(debug, "fetch remote data [uri = {}]: body is empty", uri_.uri());

source/common/http/async_client_impl.cc

@@ -257,11 +257,11 @@ AsyncRequestImpl::AsyncRequestImpl(RequestMessagePtr&& request, AsyncClientImpl&
 
 void AsyncRequestImpl::initialize() {
   child_span_->injectContext(request_->headers());
-  sendHeaders(request_->headers(), !request_->body());
-  if (request_->body()) {
+  sendHeaders(request_->headers(), request_->body().length() == 0);
+  if (request_->body().length() != 0) {
     // It's possible this will be a no-op due to a local response synchronously generated in
     // sendHeaders; guards handle this within AsyncStreamImpl.
-    sendData(*request_->body(), true);
+    sendData(request_->body(), true);
   }
   // TODO(mattklein123): Support request trailers.
 }
@@ -283,11 +283,8 @@ void AsyncRequestImpl::onHeaders(ResponseHeaderMapPtr&& headers, bool) {
 }
 
 void AsyncRequestImpl::onData(Buffer::Instance& data, bool) {
-  if (!response_->body()) {
-    response_->body() = std::make_unique<Buffer::OwnedImpl>();
-  }
   streamInfo().addBytesReceived(data.length());
-  response_->body()->move(data);
+  response_->body().move(data);
 }
 
 void AsyncRequestImpl::onTrailers(ResponseTrailerMapPtr&& trailers) {

source/common/http/async_client_impl.h

@@ -468,7 +468,7 @@ class AsyncRequestImpl final : public AsyncClient::Request,
     // The request is already fully buffered. Note that this is only called via the async client's
     // internal use of the router filter which uses this function for buffering.
   }
-  const Buffer::Instance* decodingBuffer() override { return request_->body().get(); }
+  const Buffer::Instance* decodingBuffer() override { return &request_->body(); }
   void modifyDecodingBuffer(std::function<void(Buffer::Instance&)>) override {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
   }

source/common/http/message_impl.h

@@ -24,22 +24,16 @@ class MessageImpl : public Message<HeadersInterfaceType, TrailersInterfaceType>
 
   // Http::Message
   HeadersInterfaceType& headers() override { return *headers_; }
-  Buffer::InstancePtr& body() override { return body_; }
+  Buffer::Instance& body() override { return body_; }
   TrailersInterfaceType* trailers() override { return trailers_.get(); }
   void trailers(std::unique_ptr<TrailersInterfaceType>&& trailers) override {
     trailers_ = std::move(trailers);
   }
-  std::string bodyAsString() const override {
-    if (body_) {
-      return body_->toString();
-    } else {
-      return "";
-    }
-  }
+  std::string bodyAsString() const override { return body_.toString(); }
 
 private:
   std::unique_ptr<HeadersInterfaceType> headers_;
-  Buffer::InstancePtr body_;
+  Buffer::OwnedImpl body_;
   std::unique_ptr<TrailersInterfaceType> trailers_;
 };
 

source/common/router/router.cc

@@ -741,7 +741,7 @@ void Filter::maybeDoShadowing() {
     Http::RequestMessagePtr request(new Http::RequestMessageImpl(
         Http::createHeaderMap<Http::RequestHeaderMapImpl>(*downstream_headers_)));
     if (callbacks_->decodingBuffer()) {
-      request->body() = std::make_unique<Buffer::OwnedImpl>(*callbacks_->decodingBuffer());
+      request->body().add(*callbacks_->decodingBuffer());
     }
     if (downstream_trailers_) {
       request->trailers(Http::createHeaderMap<Http::RequestTrailerMapImpl>(*downstream_trailers_));

source/extensions/common/aws/signer_impl.cc

@@ -82,8 +82,8 @@ std::string SignerImpl::createContentHash(Http::RequestMessage& message, bool si
     return SignatureConstants::get().HashedEmptyString;
   }
   auto& crypto_util = Envoy::Common::Crypto::UtilitySingleton::get();
-  const auto content_hash = message.body()
-                                ? Hex::encode(crypto_util.getSha256Digest(*message.body()))
+  const auto content_hash = message.body().length() > 0
+                                ? Hex::encode(crypto_util.getSha256Digest(message.body()))
                                 : SignatureConstants::get().HashedEmptyString;
   return content_hash;
 }

source/extensions/filters/common/ext_authz/ext_authz_http_impl.cc

@@ -258,8 +258,7 @@ void RawHttpClientImpl::check(RequestCallbacks& callbacks, Event::Dispatcher& di
   Http::RequestMessagePtr message =
       std::make_unique<Envoy::Http::RequestMessageImpl>(std::move(headers));
   if (request_length > 0) {
-    message->body() =
-        std::make_unique<Buffer::OwnedImpl>(request.attributes().request().http().body());
+    message->body().add(request.attributes().request().http().body());
   }
 
   const std::string& cluster = config_->cluster();

source/extensions/filters/http/common/jwks_fetcher.cc

@@ -69,9 +69,8 @@ class JwksFetcherImpl : public JwksFetcher,
     const uint64_t status_code = Http::Utility::getResponseStatus(response->headers());
     if (status_code == enumToInt(Http::Code::OK)) {
       ENVOY_LOG(debug, "{}: fetch pubkey [uri = {}]: success", __func__, uri_->uri());
-      if (response->body()) {
-        const auto len = response->body()->length();
-        const auto body = std::string(static_cast<char*>(response->body()->linearize(len)), len);
+      if (response->body().length() != 0) {
+        const auto body = response->bodyAsString();
         auto jwks =
             google::jwt_verify::Jwks::createFrom(body, google::jwt_verify::Jwks::Type::JWKS);
         if (jwks->getStatus() == google::jwt_verify::Status::Ok) {

source/extensions/filters/http/lua/lua_filter.cc

@@ -135,7 +135,7 @@ Http::AsyncClient::Request* makeHttpCall(lua_State* state, Filter& filter,
   }
 
   if (body != nullptr) {
-    message->body() = std::make_unique<Buffer::OwnedImpl>(body, body_size);
+    message->body().add(body, body_size);
     message->headers().setContentLength(body_size);
   }
 
@@ -348,9 +348,9 @@ void StreamHandleWrapper::onSuccess(const Http::AsyncClient::Request&,
   });
 
   // TODO(mattklein123): Avoid double copy here.
-  if (response->body() != nullptr) {
+  if (response->body().length() > 0) {
     lua_pushlstring(coroutine_.luaState(), response->bodyAsString().data(),
-                    response->body()->length());
+                    response->body().length());
   } else {
     lua_pushnil(coroutine_.luaState());
   }
@@ -385,7 +385,7 @@ void StreamHandleWrapper::onFailure(const Http::AsyncClient::Request& request,
       new Http::ResponseMessageImpl(Http::createHeaderMap<Http::ResponseHeaderMapImpl>(
           {{Http::Headers::get().Status,
             std::to_string(enumToInt(Http::Code::ServiceUnavailable))}})));
-  response_message->body() = std::make_unique<Buffer::OwnedImpl>("upstream failure");
+  response_message->body().add("upstream failure");
   onSuccess(request, std::move(response_message));
 }
 

source/extensions/filters/http/oauth2/oauth_client.cc

@@ -39,8 +39,6 @@ void OAuth2ClientImpl::asyncGetAccessToken(const std::string& auth_code,
   Http::RequestMessagePtr request = createPostRequest();
   const std::string body = fmt::format(GetAccessTokenBodyFormatString, auth_code, encoded_client_id,
                                        encoded_secret, encoded_cb_url);
-  request->body() = std::make_unique<Buffer::OwnedImpl>(body);
-
   ENVOY_LOG(debug, "Dispatching OAuth request for access token.");
   dispatchRequest(std::move(request));
 

source/extensions/filters/http/squash/squash_filter.cc

@@ -148,7 +148,7 @@ Http::FilterHeadersStatus SquashFilter::decodeHeaders(Http::RequestHeaderMap& he
   request->headers().setReferencePath(POST_ATTACHMENT_PATH);
   request->headers().setReferenceHost(SERVER_AUTHORITY);
   request->headers().setReferenceMethod(Http::Headers::get().MethodValues.Post);
-  request->body() = std::make_unique<Buffer::OwnedImpl>(config_->attachmentJson());
+  request->body().add(config_->attachmentJson());
 
   is_squashing_ = true;
   in_flight_request_ =

source/extensions/tracers/datadog/datadog_tracer_impl.cc

@@ -95,9 +95,7 @@ void TraceReporter::flushTraces() {
       message->headers().setReferenceKey(lower_case_headers_.at(h.first), h.second);
     }
 
-    Buffer::InstancePtr body(new Buffer::OwnedImpl());
-    body->add(encoder_->payload());
-    message->body() = std::move(body);
+    message->body().add(encoder_->payload());
     ENVOY_LOG(debug, "submitting {} trace(s) to {} with payload size {}", pendingTraces,
               encoder_->path(), encoder_->payload().size());
 

source/extensions/tracers/lightstep/lightstep_tracer_impl.cc

@@ -21,17 +21,16 @@ namespace Extensions {
 namespace Tracers {
 namespace Lightstep {
 
-static Buffer::InstancePtr serializeGrpcMessage(const lightstep::BufferChain& buffer_chain) {
-  Buffer::InstancePtr body(new Buffer::OwnedImpl());
+static void serializeGrpcMessage(const lightstep::BufferChain& buffer_chain,
+                                 Buffer::Instance& body) {
   auto size = buffer_chain.num_bytes();
   Buffer::RawSlice iovec;
-  body->reserve(size, &iovec, 1);
+  body.reserve(size, &iovec, 1);
   ASSERT(iovec.len_ >= size);
   iovec.len_ = size;
   buffer_chain.CopyOut(static_cast<char*>(iovec.mem_), size);
-  body->commit(&iovec, 1);
-  Grpc::Common::prependGrpcFrameHeader(*body);
-  return body;
+  body.commit(&iovec, 1);
+  Grpc::Common::prependGrpcFrameHeader(body);
 }
 
 static std::vector<lightstep::PropagationMode>
@@ -127,7 +126,7 @@ void LightStepDriver::LightStepTransporter::Send(std::unique_ptr<lightstep::Buff
   Http::RequestMessagePtr message = Grpc::Common::prepareHeaders(
       driver_.cluster(), lightstep::CollectorServiceFullName(), lightstep::CollectorMethodName(),
       absl::optional<std::chrono::milliseconds>(timeout));
-  message->body() = serializeGrpcMessage(*report);
+  serializeGrpcMessage(*report, message->body());
 
   if (collector_cluster_.exists()) {
     active_report_ = std::move(report);

source/extensions/tracers/zipkin/zipkin_tracer_impl.cc

@@ -186,9 +186,7 @@ void ReporterImpl::flushSpans() {
             ? Http::Headers::get().ContentTypeValues.Protobuf
             : Http::Headers::get().ContentTypeValues.Json);
 
-    Buffer::InstancePtr body = std::make_unique<Buffer::OwnedImpl>();
-    body->add(request_body);
-    message->body() = std::move(body);
+    message->body().add(request_body);
 
     const uint64_t timeout =
         driver_.runtime().snapshot().getInteger("tracing.zipkin.request_timeout", 5000U);

test/common/config/datasource_test.cc

@@ -244,7 +244,7 @@ TEST_F(AsyncDataSourceTest, LoadRemoteDataSourceSuccessIncorrectSha256) {
                  const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
     Http::ResponseMessagePtr response(new Http::ResponseMessageImpl(
         Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
-    response->body() = std::make_unique<Buffer::OwnedImpl>(body);
+    response->body().add(body);
 
     callbacks.onSuccess(request_, std::move(response));
     return nullptr;
@@ -289,7 +289,7 @@ TEST_F(AsyncDataSourceTest, LoadRemoteDataSourceSuccess) {
                  const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
     Http::ResponseMessagePtr response(new Http::ResponseMessageImpl(
         Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
-    response->body() = std::make_unique<Buffer::OwnedImpl>(body);
+    response->body().add(body);
 
     callbacks.onSuccess(request_, std::move(response));
     return nullptr;
@@ -371,7 +371,7 @@ TEST_F(AsyncDataSourceTest, DatasourceReleasedBeforeFetchingData) {
                    const Http::AsyncClient::RequestOptions&) -> Http::AsyncClient::Request* {
       Http::ResponseMessagePtr response(new Http::ResponseMessageImpl(
           Http::ResponseHeaderMapPtr{new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
-      response->body() = std::make_unique<Buffer::OwnedImpl>(body);
+      response->body().add(body);
 
       callbacks.onSuccess(request_, std::move(response));
       return nullptr;
@@ -446,7 +446,7 @@ TEST_F(AsyncDataSourceTest, LoadRemoteDataSourceWithRetry) {
                     Http::ResponseMessagePtr response(
                         new Http::ResponseMessageImpl(Http::ResponseHeaderMapPtr{
                             new Http::TestResponseHeaderMapImpl{{":status", "200"}}}));
-                    response->body() = std::make_unique<Buffer::OwnedImpl>(body);
+                    response->body().add(body);
 
                     callbacks.onSuccess(request_, std::move(response));
                     return nullptr;

test/common/config/http_subscription_impl_test.cc

@@ -32,7 +32,7 @@ TEST_F(HttpSubscriptionImplTest, BadJsonRecovery) {
   Http::ResponseHeaderMapPtr response_headers{
       new Http::TestResponseHeaderMapImpl{{":status", "200"}}};
   Http::ResponseMessagePtr message{new Http::ResponseMessageImpl(std::move(response_headers))};
-  message->body() = std::make_unique<Buffer::OwnedImpl>(";!@#badjso n");
+  message->body().add(";!@#badjso n");
   EXPECT_CALL(random_gen_, random()).WillOnce(Return(0));
   EXPECT_CALL(*timer_, enableTimer(_, _));
   EXPECT_CALL(callbacks_,

test/common/config/http_subscription_test_harness.h

@@ -139,7 +139,7 @@ class HttpSubscriptionTestHarness : public SubscriptionTestHarness {
     Http::ResponseHeaderMapPtr response_headers{
         new Http::TestResponseHeaderMapImpl{{":status", response_code}}};
     Http::ResponseMessagePtr message{new Http::ResponseMessageImpl(std::move(response_headers))};
-    message->body() = std::make_unique<Buffer::OwnedImpl>(response_json);
+    message->body().add(response_json);
     const auto decoded_resources =
         TestUtility::decodeResources<envoy::config::endpoint::v3::ClusterLoadAssignment>(
             response_pb, "cluster_name");