added server request params in filter (#69)

* added server request params in filter * added server request params in filter * null checks * added plugin for lombok * used lombok old * removed imports * spelling * updated lombok * not using lombok * using lombok and updated filters interface * renamed function * updated author * making getinstance abstract * added getinstance * added getinstance * added getinstance * added getinstance * added getinstance * renamed function and updated filters * updated stream in FilterInterceptor * updated stream in FilterInterceptor * added comment * added oncancel to handle exception * logging oncancel calls * refactored filters * refactored filters * using identity instead of getinstance * cleanup --------- Co-authored-by: ajay.jalgaonkar <[email protected]>
flipkart-incubator · Jul 10, 2024 · 4a3cf0e · 4a3cf0e
1 parent 625d21c
commit 4a3cf0e
Show file tree

Hide file tree

Showing 10 changed files with 101 additions and 25 deletions.
diff --git a/build.gradle b/build.gradle
@@ -92,7 +92,7 @@ subprojects {
                 hibernate_validator     : 'org.hibernate:hibernate-validator:5.4.1.Final',
                 javax_el                : 'org.glassfish:javax.el:3.0.1-b08',
                 opentracing             : 'io.opentracing:opentracing-api:0.31.0',
-                lombok                  : 'org.projectlombok:lombok:1.18.2',
+                lombok                  : 'org.projectlombok:lombok:1.18.30',
                 grpc_stub               : "io.grpc:grpc-stub:${grpcVersion}",
                 hystrix_core            : "com.netflix.hystrix:hystrix-core:1.5.12",
                 hystrix_metrics_stream  : "com.netflix.hystrix:hystrix-metrics-event-stream:1.5.12",

diff --git a/core/build.gradle b/core/build.gradle
@@ -1,6 +1,7 @@
 plugins {
     id 'java-library'
     id 'maven-publish'
+    id "io.freefair.lombok" version "8.6"
 }
 
 tasks.withType(GenerateModuleMetadata) {
@@ -64,6 +65,7 @@ dependencies {
     implementation 'io.prometheus:prometheus-metrics-exporter-common:1.2.0'
     implementation libraries.grpc_stub
     implementation libraries.grpc_services
+    implementation libraries.lombok
 
 
     testImplementation libraries.junit4

diff --git a/core/src/main/java/com/flipkart/gjex/core/filter/Filter.java b/core/src/main/java/com/flipkart/gjex/core/filter/Filter.java
@@ -16,7 +16,6 @@
 package com.flipkart.gjex.core.filter;
 
 import com.google.protobuf.GeneratedMessageV3;
-
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
@@ -30,17 +29,23 @@
  * @param <Res> Proto V3 message
  */
 public interface Filter<Req extends GeneratedMessageV3, Res extends GeneratedMessageV3> {
-	
+
 	/** Lifecycle methods for initializing and cleaning up resources used by this Filter*/
 	default void init() {}
 	default void destroy() {}
-
+
+	/**
+	 * Function for creating an instance of this {@link Filter}
+	 * Use only this function to get the {@link Filter} instance
+	 */
+	Filter<Req, Res> getInstance();
+
 	/**
 	 * Call-back to process Request headers and Filter out processing of the next incoming Request Proto V3 body/message. 
 	 * @param requestHeaders Request Headers
 	 * @throws StatusRuntimeException thrown with suitable {@link Status} to indicate reason for failing the request
 	 */
-	default void doFilterRequest(Metadata requestHeaders) throws StatusRuntimeException{}
+	default void doFilterRequest(ServerRequestParams serverRequestParams, Metadata requestHeaders) throws StatusRuntimeException{}
 
 	/**
 	 * Call-back to decorate or inspect the Reauest Proto V3 body/message. This Filter cannot fail processing of the Request body and hence there is no support for indicating failure.

diff --git a/core/src/main/java/com/flipkart/gjex/core/filter/ServerRequestParams.java b/core/src/main/java/com/flipkart/gjex/core/filter/ServerRequestParams.java
@@ -0,0 +1,19 @@
+package com.flipkart.gjex.core.filter;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+/**
+ * Parameters for passing to filters
+ * @author ajay.jalgaonkar
+ *
+ */
+
+@AllArgsConstructor
+@Getter
+@Builder
+public class ServerRequestParams {
+  private String clientIp;
+  private String methodName;
+}
diff --git a/examples/build.gradle b/examples/build.gradle
@@ -45,7 +45,7 @@ dependencies {
     implementation 'org.glassfish:javax.el:3.0.1-b08'
     implementation 'io.reactivex.rxjava2:rxjava:2.2.0'
     implementation 'com.fasterxml.jackson.core:jackson-annotations:2.16.1'
-    implementation 'org.projectlombok:lombok:1.18.6'
+    implementation 'org.projectlombok:lombok:1.18.30'
     implementation 'org.glassfish.jersey.containers:jersey-container-servlet:2.6'
     implementation 'com.fasterxml.jackson.jaxrs:jackson-jaxrs-json-provider:2.9.7'
 }

diff --git a/examples/src/main/java/com/flipkart/gjex/examples/helloworld/filter/AuthFilter.java b/examples/src/main/java/com/flipkart/gjex/examples/helloworld/filter/AuthFilter.java
@@ -18,6 +18,7 @@
 import javax.inject.Named;
 
 import com.flipkart.gjex.core.filter.Filter;
+import com.flipkart.gjex.core.filter.ServerRequestParams;
 import com.flipkart.gjex.core.logging.Logging;
 
 import io.grpc.Metadata;
@@ -44,7 +45,12 @@ public class AuthFilter implements Filter<HelloRequest, HelloReply>, Logging {
 	private final boolean isAuth = false;
 
 	@Override
-	public void doFilterRequest(Metadata requestHeaders) throws StatusRuntimeException {
+	public Filter<HelloRequest, HelloReply> getInstance(){
+		return new AuthFilter();
+	}
+
+	@Override
+	public void doFilterRequest(ServerRequestParams serverRequestParams, Metadata requestHeaders) throws StatusRuntimeException {
 		info("Headers found in the request : " + requestHeaders.toString());
 		this.checkAuth(requestHeaders);
 	}

diff --git a/examples/src/main/java/com/flipkart/gjex/examples/helloworld/filter/LoggingFilter.java b/examples/src/main/java/com/flipkart/gjex/examples/helloworld/filter/LoggingFilter.java
@@ -15,15 +15,12 @@
  */
 package com.flipkart.gjex.examples.helloworld.filter;
 
-import javax.inject.Named;
-
 import com.flipkart.gjex.core.filter.Filter;
 import com.flipkart.gjex.core.logging.Logging;
-
 import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.Metadata;
-import io.grpc.examples.helloworld.HelloReply;
-import io.grpc.examples.helloworld.HelloRequest;
+
+import javax.inject.Named;
 
 /**
  * An implementation of the {@link Filter} interface as example that simply logs Request information
@@ -35,14 +32,19 @@ public class LoggingFilter<Req extends GeneratedMessageV3, Res extends Generated
 	/** Custom response key to indicate request was logged on the server*/
     static final Metadata.Key<String> CUSTOM_HEADER_KEY = Metadata.Key.of("request_response_logged_header_key", Metadata.ASCII_STRING_MARSHALLER);
 
+    @Override
+    public Filter<Req,Res> getInstance(){
+        return new LoggingFilter<>();
+    }
+
     @Override
     public void doProcessRequest(Req request) {
         info("Logging from filter. Request payload is : " + request.toString());
     }
 
     @Override
-    public void doProcessResponseHeaders(Metadata reponseHeaders) {
-        reponseHeaders.put(CUSTOM_HEADER_KEY, "loggedRequestResponse");
+    public void doProcessResponseHeaders(Metadata responseHeaders) {
+        responseHeaders.put(CUSTOM_HEADER_KEY, "loggedRequestResponse");
     }
 
     @Override

diff --git a/...express-template/src/main/java/com/flipkart/grpc/jexpress/filter/CreateLoggingFilter.java b/...express-template/src/main/java/com/flipkart/grpc/jexpress/filter/CreateLoggingFilter.java
@@ -15,18 +15,28 @@
  */
 package com.flipkart.grpc.jexpress.filter;
 
+
 import com.flipkart.gjex.core.filter.Filter;
 import com.flipkart.gjex.core.logging.Logging;
 import com.flipkart.grpc.jexpress.CreateRequest;
 import com.flipkart.grpc.jexpress.CreateResponse;
 import com.flipkart.grpc.jexpress.GetRequest;
 import com.flipkart.grpc.jexpress.GetResponse;
+import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.Metadata;
 
 import javax.inject.Named;
 
 @Named("CreateLoggingFilter")
-public class CreateLoggingFilter implements Filter<CreateRequest, CreateResponse>, Logging {
+public class CreateLoggingFilter<CreateRequest extends GeneratedMessageV3,CreateResponse extends GeneratedMessageV3> implements Filter<CreateRequest,
+    CreateResponse>, Logging {
+
+    public CreateLoggingFilter(){}
+
+    @Override
+    public Filter<CreateRequest,CreateResponse> getInstance(){
+        return new CreateLoggingFilter<CreateRequest,CreateResponse>();
+    }
 
     @Override
     public void doProcessRequest(CreateRequest request) {

diff --git a/grpc-jexpress-template/src/main/java/com/flipkart/grpc/jexpress/filter/GetLoggingFilter.java b/grpc-jexpress-template/src/main/java/com/flipkart/grpc/jexpress/filter/GetLoggingFilter.java
@@ -19,12 +19,20 @@
 import com.flipkart.gjex.core.logging.Logging;
 import com.flipkart.grpc.jexpress.GetRequest;
 import com.flipkart.grpc.jexpress.GetResponse;
+import com.google.protobuf.GeneratedMessageV3;
 import io.grpc.Metadata;
 
 import javax.inject.Named;
 
 @Named("GetLoggingFilter")
-public class GetLoggingFilter implements Filter<GetRequest, GetResponse>, Logging {
+public class GetLoggingFilter<GetRequest extends GeneratedMessageV3, GetResponse extends GeneratedMessageV3> implements Filter<GetRequest, GetResponse>, Logging {
+
+    public GetLoggingFilter(){}
+
+    @Override
+    public Filter<GetRequest,GetResponse> getInstance(){
+        return new GetLoggingFilter<GetRequest,GetResponse>();
+    }
 
     @Override
     public void doProcessRequest(GetRequest request) {

diff --git a/guice/src/main/java/com/flipkart/gjex/grpc/interceptor/FilterInterceptor.java b/guice/src/main/java/com/flipkart/gjex/grpc/interceptor/FilterInterceptor.java
@@ -18,6 +18,7 @@
 import com.flipkart.gjex.core.context.GJEXContext;
 import com.flipkart.gjex.core.filter.Filter;
 import com.flipkart.gjex.core.filter.MethodFilters;
+import com.flipkart.gjex.core.filter.ServerRequestParams;
 import com.flipkart.gjex.core.logging.Logging;
 import com.flipkart.gjex.core.util.Pair;
 import com.flipkart.gjex.grpc.utils.AnnotationUtils;
@@ -26,6 +27,7 @@
 import io.grpc.Context;
 import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
 import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
+import io.grpc.Grpc;
 import io.grpc.Metadata;
 import io.grpc.ServerCall;
 import io.grpc.ServerCall.Listener;
@@ -43,6 +45,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -89,21 +92,30 @@ public void registerFilters(List<Filter> filters, List<BindableService> services
     @SuppressWarnings({"rawtypes", "unchecked"})
     @Override
     public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
-        List<Filter> filters = filtersMap.get(call.getMethodDescriptor().getFullMethodName().toLowerCase());
+        List<Filter> filterReferences =
+            filtersMap.get(call.getMethodDescriptor().getFullMethodName().toLowerCase());
         Metadata forwardHeaders = new Metadata();
-        if (filters == null) {
+        if (filterReferences == null || filterReferences.isEmpty()){
             return new SimpleForwardingServerCallListener<ReqT>(next.startCall(
-                    new SimpleForwardingServerCall<ReqT, RespT>(call) {
-                    }, headers)) {
+                new SimpleForwardingServerCall<ReqT, RespT>(call) {
+                }, headers)) {
             };
         }
+        List<Filter> filters = filterReferences.stream().map(Filter::getInstance).collect(Collectors.toList());
         for (Filter filter : filters) {
             try {
-                filter.doFilterRequest(headers);
-                for (Metadata.Key key : filter.getForwardHeaderKeys()) {
-                    Object value = headers.get(key);
-                    if (value != null) {
-                        forwardHeaders.put(key, value);
+                if (filter != null) {
+                    ServerRequestParams serverRequestParams =
+                        ServerRequestParams.builder()
+                            .clientIp(call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR).toString())
+                            .methodName(call.getMethodDescriptor().getFullMethodName().toLowerCase())
+                            .build();
+                    filter.doFilterRequest(serverRequestParams, headers);
+                    for (Metadata.Key key : filter.getForwardHeaderKeys()) {
+                        Object value = headers.get(key);
+                        if (value != null) {
+                            forwardHeaders.put(key, value);
+                        }
                     }
                 }
             } catch (StatusRuntimeException se) {
@@ -163,6 +175,18 @@ public void onMessage(ReqT request) {
                     detachContext(contextWithHeaders, previous);    // detach headers from gRPC context
                 }
             }
+
+            @Override
+            public void onCancel() {
+                Context previous = attachContext(contextWithHeaders);   // attaching headers to gRPC context
+                try {
+                    super.onCancel();
+                } catch (RuntimeException ex) {
+                    handleException(call, ex);
+                } finally {
+                    detachContext(contextWithHeaders, previous);    // detach headers from gRPC context
+                }
+            }
         };
     }