grpc: Allow user defined grpc status

include/envoy/grpc/status.h

@@ -5,9 +5,11 @@ namespace Grpc {
 
 class Status {
 public:
+  using GrpcStatus = int64_t;
+
   // If this enum is changed, then the std::unordered_map in Envoy::Grpc::Utility::nameToGrpcStatus
   // located at: //source/common/access_log/grpc/status.cc must also be changed.
-  enum GrpcStatus {
+  enum WellKnownGrpcStatus {
     // The RPC completed successfully.
     Ok = 0,
     // The RPC was canceled.
@@ -44,7 +46,7 @@ class Status {
     Unauthenticated = 16,
 
     // Maximum value of valid status codes.
-    MaximumValid = Unauthenticated,
+    MaximumKnown = Unauthenticated,
 
     // This is a non-GRPC error code, indicating the status code in gRPC headers
     // was invalid.

source/common/access_log/access_log_impl.cc

@@ -240,7 +240,7 @@ bool GrpcStatusFilter::evaluate(const StreamInfo::StreamInfo& info, const Http::
   //   2. response_headers gRPC status, if it exists.
   //   3. Inferred from info HTTP status, if it exists.
   //
-  // If none of those options exist, it will default to Grpc::Status::GrpcStatus::Unknown.
+  // If none of those options exist, it will default to Grpc::Status::WellKnownGrpcStatus::Unknown.
   const std::array<absl::optional<Grpc::Status::GrpcStatus>, 3> optional_statuses = {{
       {Grpc::Common::getGrpcStatus(response_trailers)},
       {Grpc::Common::getGrpcStatus(response_headers)},
@@ -249,7 +249,7 @@ bool GrpcStatusFilter::evaluate(const StreamInfo::StreamInfo& info, const Http::
                            : absl::nullopt},
   }};
 
-  Grpc::Status::GrpcStatus status = Grpc::Status::GrpcStatus::Unknown;
+  Grpc::Status::GrpcStatus status = Grpc::Status::WellKnownGrpcStatus::Unknown;
   for (const auto& optional_status : optional_statuses) {
     if (optional_status.has_value()) {
       status = optional_status.value();

source/common/config/delta_subscription_state.cc

@@ -110,7 +110,7 @@ void DeltaSubscriptionState::handleGoodResponse(
 
 void DeltaSubscriptionState::handleBadResponse(const EnvoyException& e, UpdateAck& ack) {
   // Note that error_detail being set is what indicates that a DeltaDiscoveryRequest is a NACK.
-  ack.error_detail_.set_code(Grpc::Status::GrpcStatus::Internal);
+  ack.error_detail_.set_code(Grpc::Status::WellKnownGrpcStatus::Internal);
   ack.error_detail_.set_message(e.what());
   disableInitFetchTimeoutTimer();
   ENVOY_LOG(warn, "delta config for {} rejected: {}", type_url_, e.what());
@@ -158,7 +158,7 @@ envoy::api::v2::DeltaDiscoveryRequest
 DeltaSubscriptionState::getNextRequestWithAck(const UpdateAck& ack) {
   envoy::api::v2::DeltaDiscoveryRequest request = getNextRequestAckless();
   request.set_response_nonce(ack.nonce_);
-  if (ack.error_detail_.code() != Grpc::Status::GrpcStatus::Ok) {
+  if (ack.error_detail_.code() != Grpc::Status::WellKnownGrpcStatus::Ok) {
     // Don't needlessly make the field present-but-empty if status is ok.
     request.mutable_error_detail()->CopyFrom(ack.error_detail_);
   }

source/common/config/grpc_mux_impl.cc

@@ -201,7 +201,7 @@ void GrpcMuxImpl::onDiscoveryResponse(
           Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);
     }
     ::google::rpc::Status* error_detail = api_state_[type_url].request_.mutable_error_detail();
-    error_detail->set_code(Grpc::Status::GrpcStatus::Internal);
+    error_detail->set_code(Grpc::Status::WellKnownGrpcStatus::Internal);
     error_detail->set_message(e.what());
   }
   api_state_[type_url].request_.set_response_nonce(message->nonce());

source/common/grpc/async_client_impl.cc

@@ -64,7 +64,7 @@ AsyncStreamImpl::AsyncStreamImpl(AsyncClientImpl& parent, absl::string_view serv
 
 void AsyncStreamImpl::initialize(bool buffer_body_for_retry) {
   if (parent_.cm_.get(parent_.remote_cluster_name_) == nullptr) {
-    callbacks_.onRemoteClose(Status::GrpcStatus::Unavailable, "Cluster not available");
+    callbacks_.onRemoteClose(Status::WellKnownGrpcStatus::Unavailable, "Cluster not available");
     http_reset_ = true;
     return;
   }
@@ -74,7 +74,7 @@ void AsyncStreamImpl::initialize(bool buffer_body_for_retry) {
   stream_ = http_async_client.start(*this, options_.setBufferBodyForRetry(buffer_body_for_retry));
 
   if (stream_ == nullptr) {
-    callbacks_.onRemoteClose(Status::GrpcStatus::Unavailable, EMPTY_STRING);
+    callbacks_.onRemoteClose(Status::WellKnownGrpcStatus::Unavailable, EMPTY_STRING);
     http_reset_ = true;
     return;
   }
@@ -112,8 +112,8 @@ void AsyncStreamImpl::onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) {
     // Technically this should be
     // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
     // as given by Grpc::Utility::httpToGrpcStatus(), but the Google gRPC client treats
-    // this as GrpcStatus::Canceled.
-    streamError(Status::GrpcStatus::Canceled);
+    // this as WellKnownGrpcStatus::Canceled.
+    streamError(Status::WellKnownGrpcStatus::Canceled);
     return;
   }
   if (end_stream) {
@@ -124,24 +124,24 @@ void AsyncStreamImpl::onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) {
 void AsyncStreamImpl::onData(Buffer::Instance& data, bool end_stream) {
   decoded_frames_.clear();
   if (!decoder_.decode(data, decoded_frames_)) {
-    streamError(Status::GrpcStatus::Internal);
+    streamError(Status::WellKnownGrpcStatus::Internal);
     return;
   }
 
   for (auto& frame : decoded_frames_) {
     if (frame.length_ > 0 && frame.flags_ != GRPC_FH_DEFAULT) {
-      streamError(Status::GrpcStatus::Internal);
+      streamError(Status::WellKnownGrpcStatus::Internal);
       return;
     }
     if (!callbacks_.onReceiveMessageRaw(frame.data_ ? std::move(frame.data_)
                                                     : std::make_unique<Buffer::OwnedImpl>())) {
-      streamError(Status::GrpcStatus::Internal);
+      streamError(Status::WellKnownGrpcStatus::Internal);
       return;
     }
   }
 
   if (end_stream) {
-    streamError(Status::GrpcStatus::Unknown);
+    streamError(Status::WellKnownGrpcStatus::Unknown);
   }
 }
 
@@ -152,7 +152,7 @@ void AsyncStreamImpl::onTrailers(Http::HeaderMapPtr&& trailers) {
   const std::string grpc_message = Common::getGrpcMessage(*trailers);
   callbacks_.onReceiveTrailingMetadata(std::move(trailers));
   if (!grpc_status) {
-    grpc_status = Status::GrpcStatus::Unknown;
+    grpc_status = Status::WellKnownGrpcStatus::Unknown;
   }
   callbacks_.onRemoteClose(grpc_status.value(), grpc_message);
   cleanup();
@@ -174,7 +174,7 @@ void AsyncStreamImpl::onReset() {
   }
 
   http_reset_ = true;
-  streamError(Status::GrpcStatus::Internal);
+  streamError(Status::WellKnownGrpcStatus::Internal);
 }
 
 void AsyncStreamImpl::sendMessage(const Protobuf::Message& request, bool end_stream) {
@@ -252,7 +252,7 @@ void AsyncRequestImpl::onReceiveTrailingMetadata(Http::HeaderMapPtr&&) {}
 void AsyncRequestImpl::onRemoteClose(Grpc::Status::GrpcStatus status, const std::string& message) {
   current_span_->setTag(Tracing::Tags::get().GrpcStatusCode, std::to_string(status));
 
-  if (status != Grpc::Status::GrpcStatus::Ok) {
+  if (status != Grpc::Status::WellKnownGrpcStatus::Ok) {
     current_span_->setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);
     callbacks_.onFailure(status, message, *current_span_);
   } else if (response_ == nullptr) {

source/common/grpc/common.cc

@@ -48,16 +48,17 @@ bool Common::isGrpcResponseHeader(const Http::HeaderMap& headers, bool end_strea
   return hasGrpcContentType(headers);
 }
 
-absl::optional<Status::GrpcStatus> Common::getGrpcStatus(const Http::HeaderMap& trailers) {
+absl::optional<Status::GrpcStatus> Common::getGrpcStatus(const Http::HeaderMap& trailers,
+                                                         bool allow_user_defined) {
   const Http::HeaderEntry* grpc_status_header = trailers.GrpcStatus();
-
   uint64_t grpc_status_code;
+
   if (!grpc_status_header || grpc_status_header->value().empty()) {
     return absl::nullopt;
   }
   if (!absl::SimpleAtoi(grpc_status_header->value().getStringView(), &grpc_status_code) ||
-      grpc_status_code > Status::GrpcStatus::MaximumValid) {
-    return {Status::GrpcStatus::InvalidCode};
+      (grpc_status_code > Status::WellKnownGrpcStatus::MaximumKnown && !allow_user_defined)) {
+    return {Status::WellKnownGrpcStatus::InvalidCode};
   }
   return {static_cast<Status::GrpcStatus>(grpc_status_code)};
 }
@@ -222,7 +223,7 @@ void Common::checkForHeaderOnlyError(Http::Message& http_response) {
     return;
   }
 
-  if (grpc_status_code.value() == Status::GrpcStatus::InvalidCode) {
+  if (grpc_status_code.value() == Status::WellKnownGrpcStatus::InvalidCode) {
     throw Exception(absl::optional<uint64_t>(), "bad grpc-status header");
   }
 

source/common/grpc/common.h

@@ -45,10 +45,13 @@ class Common {
   /**
    * Returns the GrpcStatus code from a given set of trailers, if present.
    * @param trailers the trailers to parse.
+   * @param allow_user_status whether allow user defined grpc status.
+   *        if this value is false, custom grpc status is regarded as invalid status
    * @return absl::optional<Status::GrpcStatus> the parsed status code or InvalidCode if no valid
    * status is found.
    */
-  static absl::optional<Status::GrpcStatus> getGrpcStatus(const Http::HeaderMap& trailers);
+  static absl::optional<Status::GrpcStatus> getGrpcStatus(const Http::HeaderMap& trailers,
+                                                          bool allow_user_defined = false);
 
   /**
    * Returns the grpc-message from a given set of trailers, if present.

source/common/grpc/google_async_client_impl.cc

@@ -84,7 +84,7 @@ GoogleAsyncClientImpl::GoogleAsyncClientImpl(Event::Dispatcher& dispatcher,
   stub_ = stub_factory.createStub(channel);
   // Initialize client stats.
   stats_.streams_total_ = &scope_->counter("streams_total");
-  for (uint32_t i = 0; i <= Status::GrpcStatus::MaximumValid; ++i) {
+  for (uint32_t i = 0; i <= Status::WellKnownGrpcStatus::MaximumKnown; ++i) {
     stats_.streams_closed_[i] = &scope_->counter(fmt::format("streams_closed_{}", i));
   }
 }
@@ -177,7 +177,7 @@ void GoogleAsyncStreamImpl::initialize(bool /*buffer_body_for_retry*/) {
   rw_ = parent_.stub_->PrepareCall(&ctxt_, "/" + service_full_name_ + "/" + method_name_,
                                    &parent_.tls_.completionQueue());
   if (rw_ == nullptr) {
-    notifyRemoteClose(Status::GrpcStatus::Unavailable, nullptr, EMPTY_STRING);
+    notifyRemoteClose(Status::WellKnownGrpcStatus::Unavailable, nullptr, EMPTY_STRING);
     call_failed_ = true;
     return;
   }
@@ -189,12 +189,12 @@ void GoogleAsyncStreamImpl::initialize(bool /*buffer_body_for_retry*/) {
 void GoogleAsyncStreamImpl::notifyRemoteClose(Status::GrpcStatus grpc_status,
                                               Http::HeaderMapPtr trailing_metadata,
                                               const std::string& message) {
-  if (grpc_status > Status::GrpcStatus::MaximumValid || grpc_status < 0) {
+  if (grpc_status > Status::WellKnownGrpcStatus::MaximumKnown || grpc_status < 0) {
     ENVOY_LOG(error, "notifyRemoteClose invalid gRPC status code {}", grpc_status);
     // Set the grpc_status as InvalidCode but increment the Unknown stream to avoid out-of-range
     // crash..
-    grpc_status = Status::GrpcStatus::InvalidCode;
-    parent_.stats_.streams_closed_[Status::GrpcStatus::Unknown]->inc();
+    grpc_status = Status::WellKnownGrpcStatus::InvalidCode;
+    parent_.stats_.streams_closed_[Status::WellKnownGrpcStatus::Unknown]->inc();
   } else {
     parent_.stats_.streams_closed_[grpc_status]->inc();
   }
@@ -272,7 +272,7 @@ void GoogleAsyncStreamImpl::handleOpCompletion(GoogleAsyncTag::Operation op, boo
     // Early fails can be just treated as Internal.
     if (op == GoogleAsyncTag::Operation::Init ||
         op == GoogleAsyncTag::Operation::ReadInitialMetadata) {
-      notifyRemoteClose(Status::GrpcStatus::Internal, nullptr, EMPTY_STRING);
+      notifyRemoteClose(Status::WellKnownGrpcStatus::Internal, nullptr, EMPTY_STRING);
       resetStream();
       return;
     }
@@ -324,7 +324,7 @@ void GoogleAsyncStreamImpl::handleOpCompletion(GoogleAsyncTag::Operation op, boo
     auto buffer = GoogleGrpcUtils::makeBufferInstance(read_buf_);
     if (!buffer || !callbacks_.onReceiveMessageRaw(std::move(buffer))) {
       // This is basically streamError in Grpc::AsyncClientImpl.
-      notifyRemoteClose(Status::GrpcStatus::Internal, nullptr, EMPTY_STRING);
+      notifyRemoteClose(Status::WellKnownGrpcStatus::Internal, nullptr, EMPTY_STRING);
       resetStream();
       break;
     }
@@ -438,7 +438,7 @@ void GoogleAsyncRequestImpl::onRemoteClose(Grpc::Status::GrpcStatus status,
                                            const std::string& message) {
   current_span_->setTag(Tracing::Tags::get().GrpcStatusCode, std::to_string(status));
 
-  if (status != Grpc::Status::GrpcStatus::Ok) {
+  if (status != Grpc::Status::WellKnownGrpcStatus::Ok) {
     current_span_->setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);
     callbacks_.onFailure(status, message, *current_span_);
   } else if (response_ == nullptr) {

source/common/grpc/google_async_client_impl.h

@@ -118,7 +118,7 @@ struct GoogleAsyncClientStats {
   // .streams_total
   Stats::Counter* streams_total_;
   // .streams_closed_<gRPC status code>
-  std::array<Stats::Counter*, Status::GrpcStatus::MaximumValid + 1> streams_closed_;
+  std::array<Stats::Counter*, Status::WellKnownGrpcStatus::MaximumKnown + 1> streams_closed_;
 };
 
 // Interface to allow the gRPC stub to be mocked out by tests.

source/common/grpc/status.cc

@@ -8,77 +8,77 @@ Status::GrpcStatus Utility::httpToGrpcStatus(uint64_t http_response_status) {
   // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md.
   switch (http_response_status) {
   case 400:
-    return Status::GrpcStatus::Internal;
+    return Status::WellKnownGrpcStatus::Internal;
   case 401:
-    return Status::GrpcStatus::Unauthenticated;
+    return Status::WellKnownGrpcStatus::Unauthenticated;
   case 403:
-    return Status::GrpcStatus::PermissionDenied;
+    return Status::WellKnownGrpcStatus::PermissionDenied;
   case 404:
-    return Status::GrpcStatus::Unimplemented;
+    return Status::WellKnownGrpcStatus::Unimplemented;
   case 429:
   case 502:
   case 503:
   case 504:
-    return Status::GrpcStatus::Unavailable;
+    return Status::WellKnownGrpcStatus::Unavailable;
   default:
-    return Status::GrpcStatus::Unknown;
+    return Status::WellKnownGrpcStatus::Unknown;
   }
 }
 
 uint64_t Utility::grpcToHttpStatus(Status::GrpcStatus grpc_status) {
   // From https://cloud.google.com/apis/design/errors#handling_errors.
   switch (grpc_status) {
-  case Status::GrpcStatus::Ok:
+  case Status::WellKnownGrpcStatus::Ok:
     return 200;
-  case Status::GrpcStatus::Canceled:
+  case Status::WellKnownGrpcStatus::Canceled:
     // Client closed request.
     return 499;
-  case Status::GrpcStatus::Unknown:
+  case Status::WellKnownGrpcStatus::Unknown:
     // Internal server error.
     return 500;
-  case Status::GrpcStatus::InvalidArgument:
+  case Status::WellKnownGrpcStatus::InvalidArgument:
     // Bad request.
     return 400;
-  case Status::GrpcStatus::DeadlineExceeded:
+  case Status::WellKnownGrpcStatus::DeadlineExceeded:
     // Gateway Time-out.
     return 504;
-  case Status::GrpcStatus::NotFound:
+  case Status::WellKnownGrpcStatus::NotFound:
     // Not found.
     return 404;
-  case Status::GrpcStatus::AlreadyExists:
+  case Status::WellKnownGrpcStatus::AlreadyExists:
     // Conflict.
     return 409;
-  case Status::GrpcStatus::PermissionDenied:
+  case Status::WellKnownGrpcStatus::PermissionDenied:
     // Forbidden.
     return 403;
-  case Status::GrpcStatus::ResourceExhausted:
+  case Status::WellKnownGrpcStatus::ResourceExhausted:
     //  Too many requests.
     return 429;
-  case Status::GrpcStatus::FailedPrecondition:
+  case Status::WellKnownGrpcStatus::FailedPrecondition:
     // Bad request.
     return 400;
-  case Status::GrpcStatus::Aborted:
+  case Status::WellKnownGrpcStatus::Aborted:
     // Conflict.
     return 409;
-  case Status::GrpcStatus::OutOfRange:
+  case Status::WellKnownGrpcStatus::OutOfRange:
     // Bad request.
     return 400;
-  case Status::GrpcStatus::Unimplemented:
+  case Status::WellKnownGrpcStatus::Unimplemented:
     // Not implemented.
     return 501;
-  case Status::GrpcStatus::Internal:
+  case Status::WellKnownGrpcStatus::Internal:
     // Internal server error.
     return 500;
-  case Status::GrpcStatus::Unavailable:
+  case Status::WellKnownGrpcStatus::Unavailable:
     // Service unavailable.
     return 503;
-  case Status::GrpcStatus::DataLoss:
+  case Status::WellKnownGrpcStatus::DataLoss:
     // Internal server error.
     return 500;
-  case Status::GrpcStatus::Unauthenticated:
+  case Status::WellKnownGrpcStatus::Unauthenticated:
     // Unauthorized.
     return 401;
-  case Status::GrpcStatus::InvalidCode:
+  case Status::WellKnownGrpcStatus::InvalidCode:
   default:
     // Internal server error.
     return 500;

source/common/grpc/typed_async_client.h

@@ -64,7 +64,7 @@ template <typename Response> class AsyncRequestCallbacks : public RawAsyncReques
         Internal::parseMessageUntyped(std::make_unique<Response>(), std::move(response))
             .release()));
     if (!message) {
-      onFailure(Status::GrpcStatus::Internal, "", span);
+      onFailure(Status::WellKnownGrpcStatus::Internal, "", span);
       return;
     }
     onSuccess(std::move(message), span);

source/common/router/router.cc

@@ -565,7 +565,6 @@ Http::ConnectionPool::Instance* Filter::getConnPool() {
   // Choose protocol based on cluster configuration and downstream connection
   // Note: Cluster may downgrade HTTP2 to HTTP1 based on runtime configuration.
   Http::Protocol protocol = cluster_->upstreamHttpProtocol(callbacks_->streamInfo().protocol());
-
   transport_socket_options_ = Network::TransportSocketOptionsUtility::fromFilterState(
       callbacks_->streamInfo().filterState());
 

source/common/tracing/http_tracer_impl.cc

@@ -94,7 +94,7 @@ static void addGrpcTags(Span& span, const Http::HeaderMap& headers) {
   }
   absl::optional<Grpc::Status::GrpcStatus> grpc_status_code = Grpc::Common::getGrpcStatus(headers);
   // Set error tag when status is not OK.
-  if (grpc_status_code && grpc_status_code.value() != Grpc::Status::GrpcStatus::Ok) {
+  if (grpc_status_code && grpc_status_code.value() != Grpc::Status::WellKnownGrpcStatus::Ok) {
     span.setTag(Tracing::Tags::get().Error, Tracing::Tags::get().True);
   }
 }