forked from raystack/firehose
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Criteria : Enable dynamic gRPC metadata based on payload Payload is parsed using CEL expression and mapped to simple key value header Metadata could contains both static and dynamic key-value Using CEL Expression to parse the Proto payload, for details please refer https://github.com/google/cel-spec/blob/master/doc/langdef.md. The reason of using CEL expression is because it natively support proto/DynamicMessage without the need to parse it to JSON first Should use full qualified proto class name for the CEL Expression Only supports primitive value as the key and value for the metadata Reason for change : Some of EGLC users that want to migrate are expecting Firehose to have the feature to insert gRPC metadata based on the payload/event Configuration would be something like this : SINK_GRPC_METADATA=staticKey : $Payload.field_one,Payload.field_two: $Payload.fields[0].name Commits: * Initial commit * streamline Cel program pipeline * Refactor code * trim metadata key * Rename method * Add assertions for buildGrpcMetadata * Add tests * Add tests and refactor classes * Add proto for test * Add mock config * Add more test * Refactor to use existing config * Add test for empty config * Test unmapped field * Update grpc-sink.md * Rename several variables to be more clear * Add inline documentation * Refactor several CEL functionalities to the utility class * Refactor naming * Checkstyle * - Add more unit test - Add checking on initialization for type support * - Checkstyle * Use built in exception for unsupported operation * Update to classpath("org.jfrog.buildinfo:build-info-extractor-gradle:4.33.1") * Remove parentheses on classpath * - Reduce try catch block - Use annotation to assert exception in test * Bump version * Bump version
- Loading branch information
1 parent
5178fd9
commit a368c89
Showing
13 changed files
with
415 additions
and
46 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
125 changes: 125 additions & 0 deletions
125
src/main/java/com/gotocompany/firehose/proto/ProtoToMetadataMapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package com.gotocompany.firehose.proto; | ||
|
|
||
| import com.google.protobuf.Descriptors; | ||
| import com.google.protobuf.DynamicMessage; | ||
| import com.google.protobuf.InvalidProtocolBufferException; | ||
| import com.google.protobuf.Message; | ||
| import com.gotocompany.firehose.exception.DeserializerException; | ||
| import com.gotocompany.firehose.utils.CelUtils; | ||
| import dev.cel.compiler.CelCompiler; | ||
| import dev.cel.runtime.CelRuntime; | ||
| import dev.cel.runtime.CelRuntimeFactory; | ||
| import io.grpc.Metadata; | ||
| import org.apache.commons.collections.MapUtils; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.Optional; | ||
| import java.util.function.Function; | ||
| import java.util.regex.Matcher; | ||
| import java.util.regex.Pattern; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
|
|
||
| /** | ||
| * Class responsible for mapping Protobuf messages to gRPC metadata using CEL expressions. | ||
| */ | ||
| public class ProtoToMetadataMapper { | ||
|
|
||
| private static final Pattern CEL_EXPRESSION_MARKER = Pattern.compile("^\\$(.+)"); | ||
| private static final int EXACT_CEL_EXPRESSION_GROUP_INDEX = 1; | ||
|
|
||
| private final Map<String, CelRuntime.Program> celExpressionToProgramMap; | ||
| private final Map<String, String> metadataTemplate; | ||
| private final Descriptors.Descriptor descriptor; | ||
|
|
||
| /** | ||
| * Constructor for ProtoToMetadataMapper. | ||
| * | ||
| * @param descriptor the Protobuf descriptor of the message type | ||
| * @param metadataTemplate a map of metadata keys and values that may contain CEL expressions | ||
| */ | ||
| public ProtoToMetadataMapper(Descriptors.Descriptor descriptor, Map<String, String> metadataTemplate) { | ||
| this.metadataTemplate = metadataTemplate; | ||
| this.descriptor = descriptor; | ||
| this.celExpressionToProgramMap = initializeCelPrograms(); | ||
| } | ||
|
|
||
| /** | ||
| * Builds gRPC metadata from a Protobuf message in byte array format. | ||
| * | ||
| * @param message the Protobuf message as a byte array | ||
| * @return gRPC metadata | ||
| * @throws DeserializerException if the Protobuf message cannot be parsed | ||
| */ | ||
| public Metadata buildGrpcMetadata(byte[] message) { | ||
| if (MapUtils.isEmpty(metadataTemplate)) { | ||
| return new Metadata(); | ||
| } | ||
| try { | ||
| return buildGrpcMetadata(DynamicMessage.parseFrom(descriptor, message)); | ||
| } catch (InvalidProtocolBufferException e) { | ||
| throw new DeserializerException("Failed to parse protobuf message", e); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Builds gRPC metadata from a Protobuf message. | ||
| * | ||
| * @param message the Protobuf message | ||
| * @return gRPC metadata | ||
| */ | ||
| private Metadata buildGrpcMetadata(Message message) { | ||
| Metadata metadata = new Metadata(); | ||
| for (Map.Entry<String, String> entry : metadataTemplate.entrySet()) { | ||
| String updatedKey = evaluateExpression(entry.getKey(), message).toString(); | ||
| Object updatedValue = evaluateExpression(entry.getValue(), message); | ||
| metadata.put(Metadata.Key.of(updatedKey.trim(), Metadata.ASCII_STRING_MARSHALLER), updatedValue.toString()); | ||
| } | ||
| return metadata; | ||
| } | ||
|
|
||
| /** | ||
| * Evaluates a CEL expression or returns the input string if it's not a CEL expression. | ||
| * | ||
| * @param input the expression to evaluate | ||
| * @param message the Protobuf message used for evaluation | ||
| * @return the evaluated result or the original expression if not a CEL expression | ||
| */ | ||
| private Object evaluateExpression(String input, Message message) { | ||
| Matcher matcher = CEL_EXPRESSION_MARKER.matcher(input); | ||
| if (!matcher.find()) { | ||
| return input; | ||
| } | ||
| String celExpression = matcher.group(EXACT_CEL_EXPRESSION_GROUP_INDEX); | ||
| return Optional.ofNullable(celExpressionToProgramMap.get(celExpression)) | ||
| .map(program -> CelUtils.evaluate(program, message)).orElse(input); | ||
| } | ||
|
|
||
| /** | ||
| * Initializes CEL programs for the metadata template. | ||
| * | ||
| * @return a map of CEL expressions to their corresponding programs | ||
| */ | ||
| private Map<String, CelRuntime.Program> initializeCelPrograms() { | ||
| CelRuntime celRuntime = CelRuntimeFactory.standardCelRuntimeBuilder().build(); | ||
| CelCompiler celCompiler = CelUtils.initializeCelCompiler(this.descriptor); | ||
| return this.metadataTemplate.entrySet() | ||
| .stream() | ||
| .filter(entry -> Objects.nonNull(entry.getValue())) | ||
| .flatMap(entry -> Stream.of(entry.getKey(), entry.getValue())) | ||
| .map(string -> { | ||
| Matcher matcher = CEL_EXPRESSION_MARKER.matcher(string); | ||
| if (matcher.find()) { | ||
| return matcher.group(EXACT_CEL_EXPRESSION_GROUP_INDEX); | ||
| } | ||
| return null; | ||
| }) | ||
| .filter(Objects::nonNull) | ||
| .distinct() | ||
| .collect(Collectors.toMap(Function.identity(), celExpression -> | ||
| CelUtils.initializeCelProgram(celExpression, celRuntime, celCompiler, celType -> celType.kind() | ||
| .isPrimitive()))); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.