We write own gateway for Thrift API

Microservices, whatever one may say, is our everything. You can resist SOAP 2.0 for as long as you like, but sooner or later they will come for you and convert you to their faith, or you will come to them yourself and ask them to baptize themselves with fire and sword. Like any architectural solution, microservices have their drawbacks. One of them is the need for every microservice to include some kind of logic for authorizing requests from external systems or other microservices. This logic can be directly “sewn up” within microservice (and it does not matter that it is a separate library), delegated to another microservice, or it can be declared declaratively. What does it mean declaratively? For example, you can agree that a special HTTP header comes to each microservice, or some kind of data structure in which there is information about the user making the request. And the data in this structure must be unequivocally trusted. All three options have their drawbacks, but in the article we will analyze the latter. To implement it, the API Gateway design pattern is usually used:


Under the cut all the difficulties of implementing the pattern in a binary data transfer protocol.

In general, API Gateway limits the number of requests for internal services, authorizes customer requests, logs and audits, distributes requests among clients, and converts data if necessary. As an example, the usual nginx can be used. Consider the function of authorizing user requests. If the HTTP protocol is used, then it is common practice to add a certain token (no matter how we got it) to the Authorization header:

Authorization: Bearer <some token> 

On the API Gateway side, this header is somehow checked and exchanged for another header containing some knowledge about the user to whom the token was written out, for example, its identifier, and you can forward it to internal services:
')

 Customer: <id> 

Everything seems simple and clear, but the trouble is that Apache Thrift consists of several parts:

 +-------------------------------------------+ | Server | | (single-threaded, event-driven etc) | +-------------------------------------------+ | Processor | | (compiler generated) | +-------------------------------------------+ | Protocol | | (JSON, compact, binary etc) | +-------------------------------------------+ | Transport | | (raw TCP, HTTP etc) | +-------------------------------------------+ 

In general, we can not tie in a protocol or transport. You can of course choose one thing, everyone can agree that we only use HTTP, but this limits the ability to replace the transport and forces you to do some external processors / filters already inside the Thrift services themselves (after all, for them the http-headers are not native) .

It remains to use the capabilities of the protocol itself, so that in the process of passing the request through the API gateway, the external authorization token is replaced with the internal one.

Convention over configuration

So, let us have the following internal service:

 service InternalTestService { SomeReturnData getSomeData( 1: UserData userData, 2: RequestData requestData ) throws (1: SomeException e); } 

UserData is some information about the user, on whose behalf the service is called, so that the latter can understand, and whose data to pull. It is clear that such a service can not be put out. And what can? For example:

 service ExternalTestService { SomeReturnData getSomeData( 1: AuthToken authData, 2: RequestData requestData ) throws (1: SomeException e, 99: UnauthorizedException ue); } 

All the difference between the services is their first argument and exception, which is initiated in case of problems with the authorization of the request (I hope that 98 own exceptions will suffice for everyone). Our task at the gateway level is to check the authorization token and replace it with user information.

Intestines

Unfortunately, the documentation for Thrift's cat wept. Almost all the guides, including, perhaps, the best of them, do not concern the internal structure of these or other protocols. And this is understandable. In 99% of cases, the developer does not have to climb inside the protocol, but we need something.

There are three most popular protocols:

• Binary - just a binary data protocol (strings, for example, are transmitted as it is in UTF-8)
• Compact - the same binary only compact
• JSON - very original JSON

Each of the presented protocols has its own implementation, hidden behind the same API. If we consider the binary protocol, then for our service it will look like this from an API point of view:


TMessage - meta information about the message. It consists of the method name, type and order number of the method in the service. Message type may be as follows:

• CALL = 1 - incoming message
• REPLY = 2 - answer
• EXCEPTION = 3 - an error occurred during execution
• ONEWAY = 4 - the message does not require a response

Everything that is not TMessage is useful information that is wrapped in the structure of the incoming message.
All submitted protocols read the incoming byte array of data sequentially and store its current index to continue reading from the right place.

Therefore, our algorithm should be as follows:

1. Read TMessage
2. Read the beginning of the general structure of the message
3. Read the meta information about the first field in the message
4. Remember current position in byte array
5. Read token information
6. Remember current position in byte array
7. Exchange token for user data
8. Serialize user data
9. To form a new binary array of three parts:
   • From the beginning of the original message to the index from point 4
   • Byte array of user data structure
   • From the index from point 6 to the end of the original message

We write a test

We don’t go to reconnaissance without testing, all the more so in the case of a binary protocol, this is the easiest way to test the performance of your code. For the test, we need the following thrift services:

Create and fill in external service with test data:

 TMemoryBuffer externalServiceBuffer = new TMemoryBufferWithLength(1024); ExternalTestService.Client externalServiceClient = new ExternalTestService.Client(protocolFactory.getProtocol(externalServiceBuffer)); externalServiceClient.send_getSomeData( new AuthToken().setToken("sometoken").setChecksum(128), new RequestData().setSomeStringField("somevalue").setSomeIntField(8) ); 

TMemoryBufferWithLength is a specially created class that eliminates the fatal flaw in the TMemoryBuffer transport for us. The latter does not know how to give the true length of the message. Instead, you can get the length of the entire buffer, which is usually greater than the length of the message because a portion of the byte array is reserved for future data.

The send_getSomeData method serializes the message to our buffer.

We will do the same with the internal service:

 internalServiceClient.send_getSomeData( new UserData().setId("user1"), new RequestData().setSomeStringField("somevalue").setSomeIntField(8) ); 

Get the byte array of our message:

 byte[] externalServiceMessage = Arrays.copyOf( externalServiceBuffer.getArray(), externalServiceBuffer.length() ); 

We introduce a class that will translate our message from a view for an external service to a view for an internal: MessageTransalator .

 public MessageTransalator(TProtocolFactory protocolFactory, AuthTokenExchanger authTokenExchanger) { this.protocolFactory = protocolFactory; this.authTokenExchanger = authTokenExchanger; } public byte[] process(byte[] thriftBody) throws TException { //some actions } 

The implementation of the token exchange ( AuthTokenExchanger ) may be different in different projects, so we will create a separate interface:

 public interface AuthTokenExchanger<T extends TBase, U extends TBase> { T createEmptyAuthToken(); U process(T authToken) throws TException; } 

createEmptyAuthToken must return an object that represents an empty token filled with MessageTransalator . In the process method, you need to implement the exchange of the access token for user data. For our test, we use a simple implementation:

 @Override public AuthToken createEmptyAuthToken() { return new AuthToken(); } @Override public UserData process(AuthToken authToken) { if ("sometoken".equals(authToken.getToken())) { return new UserData().setId("user1"); } throw new RuntimeException("token is invalid"); } 

We write a check:

 assert.assertTrue( "Translated external message must be the same as internal message", Arrays.equals( new MessageTransalator( protocolFactory, new AuthTokenExchanger<AuthToken, UserData>() {} ).process(externalServiceMessage), internalServiceMessage ) ) 

We run tests, and nothing works. And this is good!

Green light

Implement the process method according to the algorithm:

 TProtocol protocol = createProtocol(thriftBody); int startPosition = findStartPosition(protocol); TBase userData = authTokenExchanger.process( extractAuthToken(protocol, authTokenExchanger.createEmptyAuthToken()) ); int endPosition = findEndPosition(protocol); return ArrayUtils.addAll( ArrayUtils.addAll( getSkippedPart(protocol, startPosition), serializeUserData(protocolFactory, userData) ), getAfterTokenPart(protocol, endPosition, thriftBody.length) ); 

As a protocol, we use TMemoryInputTransport , which allows you to read a message directly from the byte array passed to it.

 private TProtocol createProtocol(byte[] thriftBody) { return protocolFactory.getProtocol(new TMemoryInputTransport(thriftBody)); } 

We realize the finding of token boundaries in a byte array:

 private int findStartPosition(TProtocol protocol) throws TException { skipMessageInfo(protocol); // TMessage skipToFirstFieldData(protocol); //      return protocol.getTransport().getBufferPosition(); } private int findEndPosition(TProtocol protocol) throws TException { return protocol.getTransport().getBufferPosition(); } private void skipToFirstFieldData(TProtocol protocol) throws TException { protocol.readStructBegin(); protocol.readFieldBegin(); } private void skipMessageInfo(TProtocol protocol) throws TException { protocol.readMessageBegin(); } 

Serialize user data:

 TMemoryBufferWithLength memoryBuffer = new TMemoryBufferWithLength(1024); TProtocol protocol = protocolFactory.getProtocol(memoryBuffer); userData.write(protocol); return Arrays.copyOf(memoryBuffer.getArray(), memoryBuffer.length()); 

We run the tests, and ...

Turn on Sherlock

So, tests for Binary and Compact pass, but JSON resists. What is wrong? We're going to debug and see what kind of arrays we compare:

 //JSON   [1,"getSomeData",1,1,{"1":{"rec":{"1":{"str":"user1"}}},"2":{"rec":{"1":{"str":"somevalue"},"2":{"i32":8}}}}] //JSON  [1,"getSomeData",1,1,{"1":{"rec"{"1":{"str":"user1"}}},"2":{"rec":{"1":{"str":"somevalue"},"2":{"i32":8}}}}] 

Did not notice the difference? And she is. After the first “rec” there is not enough colon. The API is the same, and the result is different. The solution came only after a careful reading of the code of the TJSONProtocol class. The protocol contains a context that stores various delimiters on the stack when it traverses the JSON structure for reading or writing.

 TJSONProtocol.JSONBaseContext context_ = new TJSONProtocol.JSONBaseContext(); 

When reading the structure, the “:” symbol is also read, but it is not returned back, because there is no context in the object itself.

Insert the crutch into the seriaizeUserData method:

 if (protocol instanceof TJSONProtocol) { memoryBuffer.write(COLON, 0, 1); // ":" } 

We run the tests, and now everything is ok.

Throw Exceptions

We are close to the finish line. Ok, let's remember that we have to throw an exception in case the request is unsuccessful:

 service ExternalTestService { SomeReturnData getSomeData( 1: AuthToken authData, 2: RequestData requestData ) throws (1: SomeException e, 99: UnauthorizedException ue); } 

We do exception handling in a separate processError method.

 public byte[] processError(TException exception) throws Exception 

In Thrift, there are several types of exceptions that may arise as a result of calling a service:

1. TApplicationException - application level exception
2. TProtocolException - protocol related exception
3. TTransportException - exception related to message passing
4. TException is a basic exception that all other types inherit from
5. YourException extends TException - any exception that was declared in DSL

An interesting detail. You can send a TApplicationException or a custom user in the reply message to the client, in our case it is UnauthorizedException . Therefore, we must wrap any errors either in a TApplicationException or in an UnauthorizedException .

 public byte[] processError(TException exception) throws Exception { TMemoryBufferWithLength memoryBuffer = new TMemoryBufferWithLength(1024); TProtocol protocol = protocolFactory.getProtocol(memoryBuffer); try { throw exception; } catch (TApplicationException e) { writeTApplicationException(e, protocol); } catch (TProtocolException e) { writeTApplicationException(createApplicationException(e), protocol); } catch (TTransportException e) { writeTApplicationException(createApplicationException(e), protocol); } catch (TException e) { if (TException.class.equals(e.getClass())) { writeTApplicationException(createApplicationException(e), protocol); } else { writeUserDefinedException(exception, protocol); } } return Arrays.copyOf(memoryBuffer.getArray(), memoryBuffer.length()); } 

Implementing a TApplicationException entry in a response data packet is fairly simple:

 private void writeTApplicationException(TApplicationException exception, TProtocol protocol) throws TException { protocol.writeMessageBegin(new TMessage(this.methodName, TMessageType.EXCEPTION, this.seqid)); exception.write(protocol); protocol.writeMessageEnd(); } private TApplicationException createApplicationException(TException e) { return new TApplicationException(TApplicationException.INTERNAL_ERROR, e.getMessage()); } 

According to the protocol, each message has its own sequence identifier and the name of the method being called, which must be returned to the client. To do this, add new fields: seqid and methodName to our class MessageTranslator , which are filled in when reading the beginning of the message. Because of this, our class ceases to be thread-safe.

To record an arbitrary exception, more gestures are required:

 private static final String ERROR_STRUCT_NAME = "result"; private static final String ERROR_FIELD_NAME = "exception"; private static final short ERROR_FIELD_POSITION = (short) 99; private static final String WRITE_METHOD_NAME = "write"; private void writeUserDefinedException(TException exception, TProtocol protocol) throws TException, IllegalAccessException, InvocationTargetException, NoSuchMethodException { TStruct errorStruct = new TStruct(ERROR_STRUCT_NAME); TField errorField = new TField(ERROR_FIELD_NAME, TType.STRUCT, ERROR_FIELD_POSITION); protocol.writeMessageBegin(new TMessage(this.methodName, TMessageType.REPLY, this.seqid)); protocol.writeStructBegin(errorStruct); protocol.writeFieldBegin(errorField); exception.getClass().getMethod(WRITE_METHOD_NAME, TProtocol.class).invoke(exception, protocol); protocol.writeFieldEnd(); protocol.writeFieldStop(); protocol.writeStructEnd(); protocol.writeMessageEnd(); } 

The interesting thing here is that for a custom exception, the type of the return message is not TMessageType.EXCEPTION , but TMessageType.REPLY .

Now we can take the incoming message, replace the token in it and correctly respond to the client if an error occurred during the verification of the token.

Spring breaks into a bar

Ok, we made the preparation of binary packages. Now is the time to make a practical implementation on the popular framework for creating microservices. For example, on Spring Boot . It is good because, on the one hand, it is possible to find ready-made solutions for it, and on the other hand, it is easy and convenient to customize it with annotations by adding new features with two or three lines of code. For routing and processing HTTP requests, take Netflix Zuul , which is included in the Spring Cloud extension set. The scheme of Zuul's work is presented in the following image:



If it is very simple, then Netflix Zuul is a regular servlet with a chain of its own filters, which can be loaded dynamically or included in the application. Each filter adds a new behavior, and even the HTTP response record is also implemented by the filter. There are several types of filters that are executed sequentially as shown in the image above. Within each type, filters are executed in the order determined by the priority of a particular filter. Connecting Zuul to the Spring Boot application is simple (add dependencies):

 @SpringBootApplication @EnableZuulProxy public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } } 

We want the same thing, but for the API gateway, so that those who will use our solution can concentrate on the business logic of authorizing their application, and not on the problems listed in the article. To do this, create an annotation @EnableThriftGateway :

 @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited @Import(ThriftGatewayConfiguration.class) public @interface EnableThriftGateway { } 

The configuration of ThriftGatewayConfiguration will contain three bins that are created if the annotation is added to the main application class: Application .

 @Configuration public class ThriftGatewayConfiguration { @Bean @ConditionalOnMissingBean(AuthTokenExchanger.class) AuthTokenExchanger authTokenExchanger() { throw new UnsupportedOperationException("You should implement AuthTokenExchanger bean"); } @Bean @ConditionalOnMissingBean(TProtocolFactory.class) TProtocolFactory thriftProtocolFactory() { return new TBinaryProtocol.Factory(); } @Bean public AuthenticationZuulFilter authenticationZuulFilter() { return new AuthenticationZuulFilter(); } } 

The ConditionalOnMissingBean annotation prevents the creation of a default bean if the application has declared its own bean of this class. The previously created AuthTokenExchanger interface should be implemented without fail by the developer of a specific project. We cannot, for security reasons, do any default implementation, so an exception is thrown in the bean creation method. Also, you need to define the protocol used for sending thrift messages. By default, this is TBinaryProtocol , but you can always use the one you need for a project by redefining the protocol factory creation bin. But the most important part of the configuration is of course the AuthenticationZuulFilter bin, which implements the business logic of the authorization layer.

 public class AuthenticationZuulFilter extends ZuulFilter { @Override public String filterType() { return "pre"; } @Override public int filterOrder() { return 6; } @Override public boolean shouldFilter() { return true; } @Override public Object run() { RequestContext ctx = RequestContext.getCurrentContext(); HttpServletRequestWrapper request = (HttpServletRequestWrapper) ctx.getRequest(); //   return null; } } 

After receiving the context objects and the HTTP request, we will create a MessageTransalator .

 MessageTransalator messageTransalator = new MessageTransalator(protocolFactory, authTokenExchanger); 

A positive scenario consists of processing the incoming data packet, recording a new packet in the requestEntity field of the request context, and specifying a new message length instead of the original one:

 byte[] processed = messageTransalator.process(request.getContentData()); ctx.set("requestEntity", new ByteArrayInputStream(processed)); ctx.setOriginContentLength(processed.length); 

If an error has occurred, it must be processed:

 ctx.setSendZuulResponse(false); ctx.setResponseDataStream(new ByteArrayInputStream(new byte[]{})); try { ctx.getResponse().getOutputStream().write(messageTransalator.processError(e)); } catch (Exception e1) { log.error("unexpected error", e1); ctx.setResponseStatusCode(HttpStatus.SC_INTERNAL_SERVER_ERROR); } 

Here we had to apply several unobvious tricks to prevent further processing of the request and attempts to send the packet to the internal service. First of all,

 ctx.setSendZuulResponse(false) 

does not allow GZIP compression of the outgoing packet. Not all thrift clients are able to survive after such repacking. And secondly,

 ctx.setResponseDataStream(new ByteArrayInputStream(new byte[]{})) 

allows you to use the original outgoing message generation filter, despite the installation in the previous paragraph of the ban on transferring data back to the client.

We put everything together

Create a new Spring Boot application and add two annotations to it, @EnableZuulProxy and @EnableThriftGateway :

 @SpringBootApplication @EnableZuulProxy @EnableThriftGateway public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } } 

We implement simple authorization logic:

 @Configuration public class AuthenticationConfiguration { @Bean AuthTokenExchanger authTokenExchanger() { return new AuthTokenExchanger<Token, TName>() { @Override public Token createEmptyAuthToken() { return new Token(); } @Override public TName process(Token authToken) throws TException { if (authToken.getValue().equals("heisours")) { return new TName("John", "Smith"); } throw new UnauthorizedException(ErrorCode.WRONG_TOKEN); } }; } } 

As you can see, if a token with the heisours value came to us, then we authorize the request, and if not, then we throw out the error. It remains only to configure Zuul:

 zuul: routes: greetings: #  URL,      path: /greetings/** #  serviceId: greetings greetings: ribbon: listOfServers: localhost:8080 # ,    greetings 

and API Gateway can be used.

Links

The basic part to convert binary packages: https://github.com/aatarasoff/thrift-api-gateway-core
Magic annotations for Spring: https://github.com/aatarasoff/spring-thrift-api-gateway
Examples: https://github.com/aatarasoff/spring-thrift-api-gateway/tree/master/examples