Skip to content

Support google.api.HttpBody in gRPC/JSON transcoding #5400

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ikhoon merged 10 commits into from
Jul 25, 2024
Merged

Support google.api.HttpBody in gRPC/JSON transcoding #5400

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0db64be
Support google.api.HttpBody in gRPC/JSON transcoding
Dogacel Jan 21, 2024
c13de0f
throw IO and return json node
Dogacel Jan 21, 2024
f213f19
continue working on supporting arbitrart HTTP data on gRPC
Dogacel Jan 21, 2024
295c2a3
fix HttpBody transcoding
Dogacel Jun 9, 2024
a8c5239
remove println
Dogacel Jun 9, 2024
d549be7
Merge branch 'main' into dogac/support-http-body-in-json-transcoding
jrhee17 Jun 14, 2024
9edd198
comment resolution
Dogacel Jun 14, 2024
cab8d06
comment resolution
Dogacel Jun 17, 2024
6478ab3
comment resolution
Dogacel Jun 28, 2024
100a429
don't close pooled data if returning the original response
Dogacel Jul 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 21 additions & 14 deletions grpc/src/main/java/com/linecorp/armeria/server/grpc/AbstractUnframedGrpcService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,8 +142,7 @@ protected void frameAndServe(
RequestHeaders grpcHeaders,
HttpData content,
CompletableFuture<HttpResponse> res,
@Nullable Function<HttpData, HttpData> responseBodyConverter,
MediaType responseContentType) {
@Nullable Function<AggregatedHttpResponse, AggregatedHttpResponse> responseConverter) {
final HttpRequest grpcRequest;
ctx.setAttr(IS_UNFRAMED_GRPC, true);
try (ArmeriaMessageFramer framer = new ArmeriaMessageFramer(
Expand Down Expand Up @@ -177,7 +176,7 @@ protected void frameAndServe(
res.completeExceptionally(t);
} else {
deframeAndRespond(ctx, framedResponse, res, unframedGrpcErrorHandler,
responseBodyConverter, responseContentType);
responseConverter);
}
}
return null;
Expand All @@ -189,8 +188,8 @@ static void deframeAndRespond(ServiceRequestContext ctx,
AggregatedHttpResponse grpcResponse,
CompletableFuture<HttpResponse> res,
UnframedGrpcErrorHandler unframedGrpcErrorHandler,
@Nullable Function<HttpData, HttpData> responseBodyConverter,
MediaType responseContentType) {
@Nullable
Function<AggregatedHttpResponse, AggregatedHttpResponse> responseConverter) {
final HttpHeaders trailers = !grpcResponse.trailers().isEmpty() ?
grpcResponse.trailers() : grpcResponse.headers();
final String grpcStatusCode = trailers.get(GrpcHeaderNames.GRPC_STATUS);
Expand Down Expand Up @@ -226,19 +225,19 @@ static void deframeAndRespond(ServiceRequestContext ctx,

final ResponseHeadersBuilder unframedHeaders = grpcResponse.headers().toBuilder();
unframedHeaders.set(GrpcHeaderNames.GRPC_STATUS, grpcStatusCode); // grpcStatusCode is 0 which is OK.
unframedHeaders.contentType(responseContentType);

final ArmeriaMessageDeframer deframer = new ArmeriaMessageDeframer(
// Max outbound message size is handled by the GrpcService, so we don't need to set it here.
Integer.MAX_VALUE);
final Subscriber<DeframedMessage> subscriber = singleSubscriber(
unframedHeaders, res, responseConverter);
grpcResponse.toHttpResponse().decode(deframer, ctx.alloc())
.subscribe(singleSubscriber(unframedHeaders, res, responseBodyConverter), ctx.eventLoop(),
SubscriptionOption.WITH_POOLED_OBJECTS);
.subscribe(subscriber, ctx.eventLoop(), SubscriptionOption.WITH_POOLED_OBJECTS);
}

static Subscriber<DeframedMessage> singleSubscriber(
ResponseHeadersBuilder unframedHeaders, CompletableFuture<HttpResponse> res,
@Nullable Function<HttpData, HttpData> responseBodyConverter) {
@Nullable Function<AggregatedHttpResponse, AggregatedHttpResponse> responseConverter) {
return new Subscriber<DeframedMessage>() {

@Override
Expand All @@ -249,12 +248,20 @@ public void onSubscribe(Subscription subscription) {
@Override
public void onNext(DeframedMessage message) {
// We know that we don't support compression, so this is always a ByteBuf.
HttpData unframedContent = HttpData.wrap(message.buf());
if (responseBodyConverter != null) {
unframedContent = responseBodyConverter.apply(unframedContent);
}
final HttpData unframedContent = HttpData.wrap(message.buf());
unframedHeaders.contentType(MediaType.JSON_UTF_8);
unframedHeaders.contentLength(unframedContent.length());
Comment thread
Dogacel marked this conversation as resolved.
Outdated
res.complete(HttpResponse.of(unframedHeaders.build(), unframedContent));

final AggregatedHttpResponse existingResponse = AggregatedHttpResponse.of(
unframedHeaders.build(),
unframedContent);

if (responseConverter != null) {
final AggregatedHttpResponse convertedResponse = responseConverter.apply(existingResponse);
res.complete(convertedResponse.toHttpResponse());
} else {
res.complete(existingResponse.toHttpResponse());
}
}

@Override
Expand Down
181 changes: 139 additions & 42 deletions grpc/src/main/java/com/linecorp/armeria/server/grpc/HttpJsonTranscodingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

import java.io.IOException;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.Base64;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
Expand All @@ -45,6 +46,7 @@
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.api.AnnotationsProto;
import com.google.api.HttpBody;
import com.google.api.HttpRule;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.CaseFormat;
Expand Down Expand Up @@ -78,6 +80,7 @@
import com.google.protobuf.Value;

import com.linecorp.armeria.common.AggregatedHttpRequest;
import com.linecorp.armeria.common.AggregatedHttpResponse;
import com.linecorp.armeria.common.HttpData;
import com.linecorp.armeria.common.HttpMethod;
import com.linecorp.armeria.common.HttpRequest;
Expand All @@ -87,6 +90,7 @@
import com.linecorp.armeria.common.QueryParams;
import com.linecorp.armeria.common.RequestHeaders;
import com.linecorp.armeria.common.RequestHeadersBuilder;
import com.linecorp.armeria.common.ResponseHeaders;
import com.linecorp.armeria.common.annotation.Nullable;
import com.linecorp.armeria.common.grpc.GrpcSerializationFormats;
import com.linecorp.armeria.common.grpc.protocol.GrpcHeaderNames;
Expand Down Expand Up @@ -195,11 +199,11 @@ static GrpcService of(GrpcService delegate, HttpJsonTranscodingOptions httpJsonT
= toRouteAndPathVariables(additionalHttpRule);
if (additionalRouteAndVariables != null) {
specs.put(additionalRouteAndVariables.getKey(),
new TranscodingSpec(order++, additionalHttpRule, methodDefinition,
serviceDesc, methodDesc, originalFields,
camelCaseFields,
additionalRouteAndVariables.getValue(),
responseBody));
new TranscodingSpec(order++, additionalHttpRule, methodDefinition,
serviceDesc, methodDesc, originalFields,
camelCaseFields,
additionalRouteAndVariables.getValue(),
responseBody));
}
}
}
Expand Down Expand Up @@ -435,7 +439,7 @@ private static String getResponseBody(List<FieldDescriptor> topLevelFields,
if (StringUtil.isNullOrEmpty(responseBody)) {
return null;
}
for (FieldDescriptor fieldDescriptor: topLevelFields) {
for (FieldDescriptor fieldDescriptor : topLevelFields) {
if (fieldDescriptor.getName().equals(responseBody)) {
return responseBody;
}
Expand All @@ -444,42 +448,102 @@ private static String getResponseBody(List<FieldDescriptor> topLevelFields,
}

@Nullable
private static Function<HttpData, HttpData> generateResponseBodyConverter(TranscodingSpec spec) {
@Nullable final String responseBody = spec.responseBody;
if (responseBody == null) {
private static MediaType getMediaTypeFromHttpBody(JsonNode jsonNode) {
final String contentType = jsonNode.get("contentType").asText();
try {
return MediaType.parse(contentType);
} catch (IllegalArgumentException e) {
logger.warn("Invalid media type in http body content_type {}.", contentType);
return null;
} else {
return httpData -> {
try (HttpData data = httpData) {

@jrhee17 jrhee17 Jun 27, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question) Can we revive the old behavior to close the httpData in case the data is pooled?

final byte[] array = data.array();
try {
final JsonNode jsonNode = mapper.readValue(array, JsonNode.class);
// we try to convert lower snake case response body to camel case
final String lowerCamelCaseResponseBody =
CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, responseBody);
final Iterator<Entry<String, JsonNode>> fields = jsonNode.fields();
while (fields.hasNext()) {
final Entry<String, JsonNode> entry = fields.next();
final String fieldName = entry.getKey();
final JsonNode responseBodyJsonNode = entry.getValue();
// try to match field name and response body
// 1. by default the marshaller would use lowerCamelCase in json field
// 2. when the marshaller use original name in .proto file when serializing messages
if (fieldName.equals(lowerCamelCaseResponseBody) ||
fieldName.equals(responseBody)) {
final byte[] bytes = mapper.writeValueAsBytes(responseBodyJsonNode);
return HttpData.wrap(bytes);
}
}
return HttpData.ofUtf8("null");
} catch (IOException e) {
logger.warn("Unexpected exception while extracting responseBody '{}' from {}",
responseBody, data, e);
return HttpData.wrap(array);
}
}
}

@Nullable
private static Function<AggregatedHttpResponse, AggregatedHttpResponse> generateResponseConverter(
TranscodingSpec spec) {
// Ignore the spec if the method is HttpBody. The response body is already in the correct format.
if (HttpBody.getDescriptor().equals(spec.methodDescriptor.getOutputType())) {
return httpResponse -> {
final JsonNode jsonNode = extractHttpBody(httpResponse.content());

// Failed to parse the JSON body, return the original response.
if (jsonNode == null) {
return httpResponse;
}

@jrhee17 jrhee17 Jun 27, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we also close the httpResponse.content() once we're sure we will be using google.api.httpBody? (Either via try blocks, or by using PooledObjects.close(httpResponse.content()))

@Dogacel Dogacel Jun 28, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I got it, I am not entirely sure how that try(...) works internally or what pooled objects are TBH. Please LMK if it looks right.

@jrhee17 jrhee17 Jul 10, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not entirely sure how that try(...) works internally or what pooled objects are TBH.

Once a gRPC message is written, it is converted to bytes via MessageMarshaller

armeria/grpc/src/main/java/com/linecorp/armeria/internal/server/grpc/AbstractServerCall.java

Line 547 in ebb0888

return responseFramer.writePayload(marshaller.serializeResponse(message));

Here the channel's ByteBufAllocator is used to allocate bytes so that no additional copies are necessary when writing over the wire.

armeria/grpc/src/main/java/com/linecorp/armeria/internal/common/grpc/GrpcMessageMarshaller.java

Line 145 in ebb0888

public ByteBuf serializeResponse(O message) throws IOException {

The response is aggregated with the pooled option, meaning don't copy the bytes

armeria/grpc/src/main/java/com/linecorp/armeria/server/grpc/AbstractUnframedGrpcService.java

Line 172 in ebb0888

grpcResponse.aggregate(AggregationOptions.usePooledObjects(ctx.alloc(), ctx.eventLoop()))

Hence, if we want to use the httpResponse as-is in the early return, we shouldn't close the httpResponse.content().

Can you remove the try block here and modify like the following?

e.g.

return httpResponse -> {
    final HttpData data = httpResponse.content();
    final JsonNode jsonNode = extractHttpBody(data);

    // Failed to parse the JSON body, return the original response.
    if (jsonNode == null) {
        return httpResponse;
    }
    PooledObjects.close(data);

    // The data field is base64 encoded.
    // https://protobuf.dev/programming-guides/proto3/#json

@Dogacel Dogacel Jul 12, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, I guess I kinda understand. What happens if we don't close the pool? I suspect the memory has to be claimed by the garbage collector which is inefficient? Or maybe there is some reference to this byte buffer left somewhere internally thus it might lead to a memory leak? (Seems unlikely)

Asking just for learning purposes 👼

@jrhee17 jrhee17 Jul 19, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is more netty territory, but my understanding is that since direct memory is not cheap to allocate/deallocate netty maintains a pool of java ByteBuffers (may be direct or heap based)

If we don't call release on the ByteBuf, the ByteBuf object itself will be garbage collected but the memory pool will continue to have a strong reference to the underlyingByteBuffer - hence a memory leak. The others may shed more light on this subject though.

// The data field is base64 encoded.
// https://protobuf.dev/programming-guides/proto3/#json
final String httpBody = jsonNode.get("data").asText();
final byte[] httpBodyBytes = Base64.getDecoder().decode(httpBody);

final ResponseHeaders newHeaders = httpResponse.headers().withMutations(builder -> {
final MediaType mediaType = getMediaTypeFromHttpBody(jsonNode);
if (mediaType != null) {
builder.contentType(mediaType);
}
builder.contentLength(httpBodyBytes.length);
});

return AggregatedHttpResponse.of(newHeaders, HttpData.wrap(httpBodyBytes));
};
}

@Nullable
final String responseBody = spec.responseBody;
if (responseBody == null) {
return null;
}

return httpResponse -> {
final HttpData convertedData = convertHttpDataForResponseBody(responseBody, httpResponse.content());
final ResponseHeaders newHeaders = httpResponse.headers().withMutations(builder -> {
builder.contentLength(convertedData.length());
});
return AggregatedHttpResponse.of(newHeaders, convertedData);
};
}

@Nullable
private static JsonNode extractHttpBody(HttpData data) {
final byte[] array = data.array();

final JsonNode jsonNode;
Comment thread
Dogacel marked this conversation as resolved.
Outdated
try {
return mapper.readValue(array, JsonNode.class);
} catch (IOException e) {
logger.warn("Unexpected exception while parsing HttpBody from {}", data, e);
return null;
}
}

private static HttpData convertHttpDataForResponseBody(String responseBody, HttpData data) {
final byte[] array = data.array();
try {
final JsonNode jsonNode = mapper.readValue(array, JsonNode.class);

// we try to convert lower snake case response body to camel case
final String lowerCamelCaseResponseBody =
CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, responseBody);
final Iterator<Entry<String, JsonNode>> fields = jsonNode.fields();
while (fields.hasNext()) {
final Entry<String, JsonNode> entry = fields.next();
final String fieldName = entry.getKey();
final JsonNode responseBodyJsonNode = entry.getValue();
// try to match field name and response body
// 1. by default the marshaller would use lowerCamelCase in json field
// 2. when the marshaller use original name in .proto file when serializing messages
if (fieldName.equals(lowerCamelCaseResponseBody) ||
fieldName.equals(responseBody)) {
final byte[] bytes = mapper.writeValueAsBytes(responseBodyJsonNode);
return HttpData.wrap(bytes);
}
}
return HttpData.ofUtf8("null");
} catch (IOException e) {
logger.warn("Unexpected exception while extracting responseBody '{}' from {}",
responseBody, data, e);
return HttpData.wrap(array);
}
}

private static final ObjectMapper mapper = JacksonUtil.newDefaultObjectMapper();
Expand Down Expand Up @@ -582,10 +646,20 @@ private HttpResponse serve0(ServiceRequestContext ctx, HttpRequest req,
} else {
try {
ctx.setAttr(FramedGrpcService.RESOLVED_GRPC_METHOD, spec.method);
// Set JSON media type (https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/grpc_json_transcoder_filter#sending-arbitrary-content)
final HttpData requestContent;

// https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/grpc_json_transcoder_filter#sending-arbitrary-content
if (HttpBody.getDescriptor().equals(spec.methodDescriptor.getInputType())) {
// Convert the HTTP request to a JSON representation of HttpBody.
requestContent = convertToHttpBody(clientRequest);
} else {
// Convert the HTTP request to gRPC JSON.
requestContent = convertToJson(ctx, clientRequest, spec);
}

frameAndServe(unwrap(), ctx, grpcHeaders.build(),
convertToJson(ctx, clientRequest, spec), responseFuture,
generateResponseBodyConverter(spec), MediaType.JSON_UTF_8);
requestContent, responseFuture,
generateResponseConverter(spec));
} catch (IllegalArgumentException iae) {
responseFuture.completeExceptionally(
HttpStatusException.of(HttpStatus.BAD_REQUEST, iae));
Expand All @@ -599,6 +673,29 @@ private HttpResponse serve0(ServiceRequestContext ctx, HttpRequest req,
return HttpResponse.of(responseFuture);
}

private static HttpData convertToHttpBody(AggregatedHttpRequest request) throws IOException {
final ObjectNode body = mapper.createObjectNode();

try (HttpData content = request.content()) {
final MediaType contentType;

@Nullable
final MediaType requestContentType = request.contentType();
if (requestContentType != null) {
contentType = requestContentType;
} else {
contentType = MediaType.OCTET_STREAM;
}

body.put("content_type", contentType.toString());
// Jackson converts byte array to base64 string. gRPC transcoding spec also returns base64 string.
// https://protobuf.dev/programming-guides/proto3/#json
body.put("data", content.array());

return HttpData.wrap(mapper.writeValueAsBytes(body));
}
}

/**
* Converts the HTTP request to gRPC JSON with the {@link TranscodingSpec}.
*/
Expand All @@ -625,7 +722,7 @@ private HttpData convertToJson(ServiceRequestContext ctx,
root = mapper.createObjectNode();
} else {
throw new IllegalArgumentException("Unexpected JSON: " +
body + ", (expected: ObjectNode or null).");
body + ", (expected: ObjectNode or null).");
}
return setParametersAndWriteJson(root, ctx, spec);
}
Expand Down
12 changes: 11 additions & 1 deletion grpc/src/main/java/com/linecorp/armeria/server/grpc/UnframedGrpcService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,17 @@

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;

import com.linecorp.armeria.common.AggregatedHttpResponse;
import com.linecorp.armeria.common.AggregationOptions;
import com.linecorp.armeria.common.HttpRequest;
import com.linecorp.armeria.common.HttpResponse;
import com.linecorp.armeria.common.HttpStatus;
import com.linecorp.armeria.common.MediaType;
import com.linecorp.armeria.common.RequestHeaders;
import com.linecorp.armeria.common.RequestHeadersBuilder;
import com.linecorp.armeria.common.ResponseHeaders;
import com.linecorp.armeria.common.SerializationFormat;
import com.linecorp.armeria.common.grpc.GrpcSerializationFormats;
import com.linecorp.armeria.common.grpc.protocol.GrpcHeaderNames;
Expand Down Expand Up @@ -153,8 +156,15 @@ public HttpResponse serve(ServiceRequestContext ctx, HttpRequest req) throws Exc
if (t != null) {
responseFuture.completeExceptionally(t);
} else {
// Add the content type to the response headers.
final Function<AggregatedHttpResponse, AggregatedHttpResponse> responseConverter =
response -> {
final ResponseHeaders headers = response.headers().withMutations(
builder -> builder.contentType(contentType));
return AggregatedHttpResponse.of(headers, response.content());
};
frameAndServe(unwrap(), ctx, grpcHeaders.build(), clientRequest.content(),
responseFuture, null, contentType);
responseFuture, responseConverter);
}
}
return null;
Expand Down
Loading