Comenzar un artículo señalando que, probablemente, el título no es el más acertado de todos los confeccionados hasta el momento, no parece la forma idílica de iniciar un escrito, pero no siempre es tan sencillo.
La idea es aprovechar la experiencia adquirida en los últimos meses en la construcción de un reverse-proxy vitaminado en Undertow, para desgranar algunos de los detalles de implementación que más chocantes pueden resultar cuando se comienza a trabajar con esta tecnología.
Caso de uso
Se desea construir un reverse-proxy en Undertow, que implemente un sencillo log de accesos, con el que mostrar los detalles de implementación de las siguientes funcionalidades:
- Non-blocking handlers
- Custom exchange attributes
- Request and response body read
- Custom error handler
- Kubernetes ready health check
Undertow
Antes de entrar de lleno en materia, una breve descripción de lo que es Undertow y sus características clave con las que sacarle el mayor provecho posible.
Undertow es un web-server ligero de alto rendimiento escrito en Java y construido sobre el API NIO (Non-blocking I/O), para la ejecución de tareas tanto bloqueantes como no bloqueantes. Es muy posible que hayas oído hablar de él o simplemente lo estés utilizando sin haberte dado cuenta, ya que puede funcionar tanto de forma standalone, modelo en el que se centra el presente artículo, o de forma embebida con Wildfly o Spring Boot.
Así, Undertow gira en base a tres conceptos básicos: workers, listeners y handlers.
XNIO worker
Los workers XNIO, una capa de abstracción sobre el API NIO mencionado previamente, son los encargados de gestionar los distintos hilos del servidor:
- WORKER_IO_THREADS: Son aquellos hilos encargados de ejecutar las tareas no bloqueantes. Es extremadamente importante que bajo ningún concepto lleven a cabo tareas bloqueantes, ya que, cómo su propio nombre deja entrever, son los encargados de gestionar la recepción y respuesta de las peticiones, y mientras se encuentren bloqueados, no podrán atender otras solicitudes. Dos hilos por núcleo de CPU es un valor por defecto razonable.
- WORKER_TASK_CORE_THREADS: Son aquellos hilos encargados de ejecutas las tareas más pesadas o bloqueantes. Diez hilos por núcleo de CPU es un valor por defecto razonable.
A estas alturas ya habréis deducido que, si se hace hincapié en describirlo, es porque es el desarrollador quien decide en qué hilo se ejecuta cada tarea. No os preocupéis, llegado el momento se detallará como implementar esta elección.
Listeners
Los listener son los encargados de gestionar las conexiones entrantes, así como el protocolo de comunicación. Undertow proporciona los siguientes listener de caja:
- HTTP / 1.1
- HTTPS
- AJP
- HTTP / 2
Así, cada vez que el servidor recibe una petición, el listener se encarga de procesar la solicitud y crear un objeto HttpServerExchange con todos los datos de esta, el cual se propagará entre los distintos handlers que formen parte de la operación.
A lo largo del artículo descubriréis que este objeto resulta de vital importancia, ya que permite acceder a los datos de la request, construir los datos de la response o compartir información en cualquier momento entre los distintos handlers.
Handlers
Los handlers son clases que implementan la interfaz io.undertow.server.HttpHandler y albergan la lógica a ejecutar, pudiéndose concatenar varios handlers para una misma operación, formando así un flujo o pipeline.
Como no podía ser de otra forma, Undertow proporciona de caja un interesante un conjunto de handlers que se recomienda encarecidamente conocer, antes de ponerse a desarrollar uno por cuenta propia.
Building reverse de proxy
Llega el momento de desgranar los detalles de implementación.
Create and configure undertow Server
El primer paso cuando se comienza a trabajar con Undertow es definir la configuración básica del servidor (workers, listener y handlers) y arrancarlo en una clase con el ya clásico método main en Java.
public static void main(String[] args) throws IOException {
undertow = Undertow.builder()
.addHttpListener(getConfiguration().getServer().getPort(), getConfiguration().getServer().getHost())
.setIoThreads(getConfiguration().getServer().getIoThreads())
.setWorkerThreads(getConfiguration().getServer().getWorkerThreads())
.setHandler(Handlers.predicates(
PredicatedHandlersParser.parse(getConfiguration().getServer().getPredicates(), null),
exchange -> Optional.of(exchange).filter(HttpServerExchange::isResponseChannelAvailable)
.ifPresent(ex -> ex.getResponseSender().send(ex.getRelativePath()))))
.build();
undertow.start();
}
Para simplificar la lectura de la configuración, se ha desarrollado una clase ConfigurationManager, la cual se encarga de leer las propiedades desde un fichero Undertow.yaml ubicado en el classpath. Evidentemente, si este fichero contuviera datos sensibles del entorno, no debería ser empaquetado junto a la aplicación, sino que debería ser añadido al classpath mediante el argumento -Xbootclasspath/.
public class ConfigurationLoader {
private static ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
private static Configuration configuration;
private static final String UNDERTOW_CONFIG_FILENAME = "/undertow.yaml";
public static Configuration getConfiguration() throws IOException {
if (configuration == null) {
configuration = mapper.readValue(ConfigurationLoader.class.getResourceAsStream(UNDERTOW_CONFIG_FILENAME),
Configuration.class);
}
return configuration;
}
}
¿Y qué pinta tiene este fichero Undertow.yaml?
server:
port: 8080
ioThreads: 12
workerThreads: 60
predicates: |
path-prefix('/health') -> {health-check(); done}
path-prefix('/') -> error-handler()
path-prefix('/') -> reverse-proxy-log ()
path-prefix('/') -> reverse-proxy({'http://localhost:8888/'})
Como seguro habréis deducido, los predicados no son más que la definición de handlers descritos previamente para cada una de las operaciones, y si, todos aquellos que tengan el mismo path-prefix se ejecutan de forma secuencial para una misma operación.
Health Check Handler
Hora de construir el primer handler, en este caso, un pequeño health check que devuelva un Status Code 200 cada vez que se realice una petición sobre el path “/health”, ideal si se pretende desplegar en Kubernetes por ejemplo.
Como ya se detalló anteriormente, todos los handlers deben implementar la interfaz io.undertow.server.HttpHandler y en consecuencia, el método handleRequest.
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
if (exchange.isInIoThread()) {
exchange.dispatch(this);
return;
}
exchange.dispatch(INSTANCE, () -> doHealthCheck(exchange).orTimeout(30, SECONDS)
.whenComplete((health, ex) -> dispatchRequest(next, exchange)));
}
Lo primero todo es comprobar el hilo en el que se va a ejecutar la petición y redireccionarlo a un hilo core. Aunque el caso de uso actual es francamente sencillo y no tiene impacto alguno en el rendimiento, siempre es recomendable ejecutar la lógica fuera de los hilos IO.
A partir de ahi, se invoca el método doHealthCheck, haciendo uso de un CompetableFuture de Java, con la idea de que se ejecute de forma asíncrona y poder concatenar o ejecutar en paralelo múltiples operaciones de una forma óptima.
De nuevo, esto tampoco es realmente necesario para el caso de uso actual, pero el objetivo del articulo pasa por detallar buenas practicas de implementación cuando se trabaja con Undertow.
private CompletableFuture<Void> doHealthCheck(HttpServerExchange exchange) {
CompletableFuture<Void> future = CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
String health = null;
exchange.getResponseHeaders().put(CONTENT_TYPE, APPLICATION_JSON);
try {
health = mapper.writeValueAsString(Health.up().build());
exchange.setStatusCode(OK);
exchange.getResponseSender().send(health);
} catch (IOException exception) {
LOGGER.error(exception);
try {
health = mapper.writeValueAsString(Health.down().build());
exchange.setStatusCode(INTERNAL_SERVER_ERROR);
exchange.getResponseSender().send(health);
} catch (IOException e) {
// Ignore
}
}
}
});
return future;
}
Como se puede observar, el exchange no solo contiene los datos de la petición, sino que también es el medio para establecer el status code y body de la respuesta.
Un pequeño inciso antes de continuar; Undertow considera que una operación está completa, si tanto la request como la response se han leído / escrito completamente. En ese momento, el flujo o pipepline se da por finalizado y no se podrán concatenar nuevos handlers tras él.
Evidentemente este caso de uso no se ve impactado, ya que es el único handler que interviene en la operación y, por tanto, el encargado de escribir el mensaje de respuesta, pero no está de mas tenerlo en cuenta.
El siguiente paso es crear una clase que implemente la interfaz HandlerBuilder, en la que definir la configuración del handler, como el nombre con el que debe ser invocado o los parámetros de entrada del mismo.
public class HealthCheckHandlerBuilder implements HandlerBuilder {
@Override
public String name() {
return "health-check";
}
@Override
public Map<String, Class<?>> parameters() {
// Set the parameter names (avoid whitespaces) and its type here
return null;
}
@Override
public Set<String> requiredParameters() {
// Return the names of the required parameters (configuration would fail if any
// of these is missing)
return null;
}
@Override
public String defaultParameter() {
// If there is only one parameter, or one required parameter, return it to make
// it easier to configure
return null;
}
@Override
public HandlerWrapper build(final Map<String, Object> config) {
// Get the configuration parameters here and parse them
// Example: String myParameter = (String)config.get("my-parameter");
// After that, return the builder
return handler -> new HealthCheckHandler(handler);
}
}
Finalmente, no queda mas que crear un fichero denominado io.undertow.server.handlers.builder.HandlerBuilder en la ruta “/src/main/resources/META-INF/services/” y añadir el HealthCheckHandlerBuilder, para que este pueda ser gestionado por Undertow.
com.mikeldeltio.undertow.handler.builder.HealthCheckHandlerBuilder
En este punto ya es posible probar lo construido, siempre y cuando antes se elimine la definición del resto de las operaciones del fichero de configuración Undertow.yaml. No olvidéis restaurarlos una vez realizadas las pruebas correspondientes.
server:
port: 8080
ioThreads: 12
workerThreads: 60
predicates: |
path-prefix('/health') -> {health-check(); done}
Reverse proxy log handler
El propósito del log handler es pintar por consola, tanto los datos de entrada de la petición recibida, así como los datos de la respuesta tras haber pasado por el reverse proxy, un handler que Undertow proporciona ya de caja.
Para ello y del mismo modo que en el health check handler, se debe crear una clase ReverseProxyLogHandler, que implemente la interfaz HttpHandler, con el correspondiente método handleRequest. Hasta aquí, nada nuevo
public class ReverseProxyLogHandler implements HttpHandler {
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
if (exchange.isInIoThread()) {
exchange.dispatch(this);
return;
}
exchange.dispatch(INSTANCE, () -> doLogRequest(exchange).orTimeout(30, SECONDS)
.whenComplete((health, ex) -> dispatchRequest(next, exchange)));
}
}
En esta ocasión, el método invocado es el doLogRequest, el cual por conveniencia del guión, se encarga de leer todos los atributos de la request y almacenarlos en el exchange, de tal forma que puedan ser utilizados después en otro método o handler.
Aclaración, el objeto HttpLogMessage no es más que una clase Java normal y corriente que contiene todos los atributos a almacenar.
private CompletableFuture<Void> doLogRequest(HttpServerExchange exchange) {
CompletableFuture<Void> future = CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
HttpLogMessage httpLogMessage = new HttpLogMessage();
httpLogMessage.setTimestamp(LocalDateTime.now().toString());
httpLogMessage.setMethod(exchange.getRequestMethod().toString());
httpLogMessage.setUrl(exchange.getRequestURL()
+ (!exchange.getQueryString().isBlank() ? "?" + exchange.getQueryString() : ""));
httpLogMessage.setRequestHeaders(exchange.getRequestHeaders().toString());
httpLogMessage.setRequestBody(readRequestBody(exchange));
setHttpLogMessage(exchange, httpLogMessage);
}
});
return future;
}
En esta función hay dos detalles importantes en los que detenerse.
El primero de ellos es el método readRequestBody, el cual se encarga de leer el cuerpo de la petición sin alterar el estado inicial del buffer, de tal forma que siga estando disponible para el siguiente handler, en este caso, el reverse proxy.
public static String readRequestBody(HttpServerExchange exchange) {
PooledByteBuffer bufferPool = exchange.getConnection().getByteBufferPool().allocate();
ByteBuffer buffer = bufferPool.getBuffer();
try {
StreamSourceChannel inChannel = exchange.getRequestChannel();
int bytesRead = inChannel.read(buffer);
while (bytesRead != -1) {
bytesRead = inChannel.read(buffer);
}
buffer.flip();
} catch (Exception e) {
if (bufferPool != null && bufferPool.isOpen()) {
IoUtils.safeClose(bufferPool);
}
e.printStackTrace();
}
Connectors.ungetRequestBytes(exchange, bufferPool);
Connectors.resetRequestChannel(exchange);
try {
return new String(getByteArrayFromByteBuffer(buffer), exchange.getRequestCharset());
} catch (UnsupportedEncodingException exception) {
LOGGER.error(exception);
return new String();
}
}
private static byte[] getByteArrayFromByteBuffer(ByteBuffer buffer) {
byte[] bytesArray = new byte[buffer.remaining()];
buffer.get(bytesArray, 0, bytesArray.length);
buffer.rewind();
return bytesArray;
}
El segundo es el método setHttpLogMessage, el cual se encarga de almacenar el objeto HttpLogMessage en el exchange.
Para ello, se simula el comportamiento de los ExchangeAttributes mediante un objeto HttpLogMessageAtributte, pero sin llegar a extender de dicha clase, lo que a la postre permite guardar datos no primitivos o complejos.
Al no tratarse de un atributo del exchange como tal, tampoco es necesario registrarlo en el fichero io.undertow.attribute.ExchangeAttributeBuilder de la ruta “/src/main/resources/META-INF/services/“.
public class HttpLogMessageAtributte {
public final static AttachmentKey<HttpLogMessage> HTTP_LOG_MSG_ATTACHMENT_KEY = AttachmentKey
.create(HttpLogMessage.class);
public static HttpLogMessage getHttpLogMessage(HttpServerExchange exchange) {
return exchange.getAttachment(HTTP_LOG_MSG_ATTACHMENT_KEY);
}
public static HttpLogMessage setHttpLogMessage(HttpServerExchange exchange, HttpLogMessage httpLogMessage) {
return exchange.putAttachment(HTTP_LOG_MSG_ATTACHMENT_KEY, httpLogMessage);
}
}
Llegados a este punto, se dispone de toda la información de la request correctamente almacenada en el exchange y lista para ser explotada una vez se reciba de la respuesta.
El problema reside en que el reverse-proxy-handler de Undertow no está diseñado para que puedan concatenarse nuevos handlers tras él, al fin y al cabo, debe escribir la respuesta obtenida del servidor invocado.
@Override
public void completed(final ClientExchange result) {
final ClientResponse response = result.getResponse();
if(log.isDebugEnabled()) {
log.debugf("Received response %s for request %s for exchange %s", response, result.getRequest(), exchange);
}
final HeaderMap inboundResponseHeaders = response.getResponseHeaders();
final HeaderMap outboundResponseHeaders = exchange.getResponseHeaders();
exchange.setStatusCode(response.getResponseCode());
copyHeaders(outboundResponseHeaders, inboundResponseHeaders);
if (exchange.isUpgrade()) {
exchange.upgradeChannel(new HttpUpgradeListener() {
@Override
public void handleUpgrade(StreamConnection streamConnection, HttpServerExchange exchange) {
if(log.isDebugEnabled()) {
log.debugf("Upgraded request %s to for exchange %s", result.getRequest(), exchange);
}
StreamConnection clientChannel = null;
try {
clientChannel = result.getConnection().performUpgrade();
final ClosingExceptionHandler handler = new ClosingExceptionHandler(streamConnection, clientChannel);
Transfer.initiateTransfer(clientChannel.getSourceChannel(), streamConnection.getSinkChannel(), ChannelListeners.closingChannelListener(), ChannelListeners.writeShutdownChannelListener(ChannelListeners.<StreamSinkChannel>flushingChannelListener(ChannelListeners.closingChannelListener(), ChannelListeners.closingChannelExceptionHandler()), ChannelListeners.closingChannelExceptionHandler()), handler, handler, result.getConnection().getBufferPool());
Transfer.initiateTransfer(streamConnection.getSourceChannel(), clientChannel.getSinkChannel(), ChannelListeners.closingChannelListener(), ChannelListeners.writeShutdownChannelListener(ChannelListeners.<StreamSinkChannel>flushingChannelListener(ChannelListeners.closingChannelListener(), ChannelListeners.closingChannelExceptionHandler()), ChannelListeners.closingChannelExceptionHandler()), handler, handler, result.getConnection().getBufferPool());
} catch (IOException e) {
IoUtils.safeClose(streamConnection, clientChannel);
}
}
});
}
final IoExceptionHandler handler = new IoExceptionHandler(exchange, result.getConnection());
Transfer.initiateTransfer(result.getResponseChannel(), exchange.getResponseChannel(), ChannelListeners.closingChannelListener(), new HTTPTrailerChannelListener(result, exchange, exchange, proxyClientHandler, idempotentPredicate), handler, handler, exchange.getConnection().getByteBufferPool());
}
La diabólica solución pasa por realizar un wrapper sobre la respuesta que permita ejecutar código antes de que esta sea escrita y devuelva la respuesta al usuario.
Para ello, lo primero es actualizar el método handleRequest y añadir un ResponseWrapper.
@Override
public void handleRequest(HttpServerExchange exchange) throws Exception {
if (exchange.isInIoThread()) {
exchange.dispatch(this);
return;
}
exchange.addResponseWrapper(new ConduitWrapper<StreamSinkConduit>() {
@Override
public StreamSinkConduit wrap(ConduitFactory<StreamSinkConduit> factory, HttpServerExchange exchange) {
return new ReverseProxyResponseStreamSinkConduit(factory.create(), exchange);
}
});
exchange.dispatch(INSTANCE, () -> doLogRequest(exchange).orTimeout(30, SECONDS)
.whenComplete((health, ex) -> dispatchRequest(next, exchange)));
}
El siguiente paso es construir el ReverseProxyResponseStreamSinkConduit, el cual se encarga de leer el contenido de la respuesta e invocar la función con la lógica de negocio a ejecutar, antes de escribir la respuesta.
Al igual que cuando se leyó el cuerpo de la request, es vital no alterar el estado del buffer de la response, al menos si se quiere devolver después el contenido usuario. Para ello, se ha optado por copiar su contenido, byte a byte.
public class ReverseProxyResponseStreamSinkConduit extends AbstractStreamSinkConduit<StreamSinkConduit> {
private final HttpServerExchange exchange;
private boolean processedResponse = false;
private byte[] bufferedResponse = new byte[] {};
public ReverseProxyResponseStreamSinkConduit(StreamSinkConduit next, HttpServerExchange exchange) {
super(next);
this.exchange = exchange;
}
@Override
public int write(ByteBuffer src) throws IOException {
if (!processedResponse) {
byte[] response = new byte[src.remaining()];
src.get(response);
int alreadyWrittenSize = bufferedResponse.length;
bufferedResponse = ByteBuffer.allocate(bufferedResponse.length + response.length).put(bufferedResponse)
.put(response).array();
if (exchange.getResponseContentLength() == bufferedResponse.length) {
doLogResponse(exchange, bufferedResponse);
processedResponse = true;
return super.write(ByteBuffer.wrap(bufferedResponse)) - alreadyWrittenSize;
} else {
return bufferedResponse.length - alreadyWrittenSize;
}
}
return super.write(src);
}
@Override
public void terminateWrites() throws IOException {
if (!processedResponse) {
doLogResponse(exchange, bufferedResponse);
processedResponse = true;
}
super.terminateWrites();
}
}
El motivo por el que se comprueba tanto en el método writte como en el terminateWrites si debe ejecutarse el tratamiento de la respuesta, es que si el reverse-proxy no devuelve un body en la respuesta, no se ejecuta el el método writte y, por tanto, el tratamiento debe llevarse a cabo en el terminateWrites.
Por el contrario, si el reverse proxy devuelve un body en la respuesta, el tratamiento de esta debe llevarse a cabo en el método writte, ya que para cuando se ejecuta el método terminateWrites, se ha devuelto la respuesta al usuario. Dado que la respuesta puede estar dividida en varios paquetes, se espera a tener el mensaje completo, en base a la cabecera “Content-Length“. Si la respuesta es mayor
Finalmente, no queda más que implementar el método doLogResponse en el ReverseProxyLogHandler, el cual recupera los datos de la request almacenados en el exchange, los enriquece con los datos de la response y los pinta por consola.
public static void doLogResponse(HttpServerExchange exchange, byte[] bufferedResponse) {
try {
HttpLogMessage httpLogMessage = getHttpLogMessage(exchange);
httpLogMessage.setStatusCode(exchange.getStatusCode());
httpLogMessage.setResponseHeaders(exchange.getResponseHeaders().toString());
httpLogMessage.setResponseBody(new String(bufferedResponse, exchange.getResponseCharset()));
LOGGER.info(mapper.writeValueAsString(httpLogMessage));
} catch (IOException exception) {
LOGGER.error(exception);
}
}
Antes de ejecutar la aplicación, no olvidéis definir un ReverseProxyLogHandlerBuilder y añadirlo al fichero io.undertow.server.handlers.builder.HandlerBuilder, de la misma forma que se llevo a cabo para el health check handler.
Error handler
El ultimo handler a construir, pero no por ello menos importante, es el encargado de capturar los errores no esperados y devolver una respuesta adecuada al usuario.
Para ello y del mismo modo que en el resto de handler, se debe crear una clase ErrorgHandler, que implemente la interfaz HttpHandler, con el correspondiente método handleRequest. Ahora bien, a diferencia del resto de ocasiones, se define un mensaje de respuesta por defecto a utilizar cuando se produce una excepción no controlada en alguno de los handlers que se ejecute a posteriore. De ahí a que sea el primer handler definido para el path “/” en el fichero de configuración de Undertow.yaml.
@Override
public void handleRequest(final HttpServerExchange exchange) throws Exception {
exchange.addDefaultResponseListener(new DefaultResponseListener() {
@Override
public boolean handleDefaultResponse(final HttpServerExchange exchange) {
if (!exchange.isResponseChannelAvailable()) {
return false;
}
Throwable exception = exchange.getAttachment(DefaultResponseListener.EXCEPTION);
ErrorMessage errorMessage = new ErrorMessage();
errorMessage.setTimestamp(LocalDateTime.now().format(ofPattern(DATE_TIME_FORMAT)).toString());
errorMessage.setDescription(StatusCodes.INTERNAL_SERVER_ERROR_STRING);
errorMessage.setMessage(exception != null ? exception.getMessage() : "");
String errorMessageJson = null;
try {
errorMessageJson = new ObjectMapper().writeValueAsString(errorMessage);
} catch (IOException e) {
LOGGER.error("Error while serializing Error Message {}", e);
errorMessageJson = "{}";
}
exchange.getResponseHeaders().put(Headers.CONTENT_TYPE, APPLICATION_JSON);
exchange.getResponseHeaders().put(Headers.CONTENT_LENGTH, errorMessageJson.length());
exchange.setStatusCode(500);
exchange.getResponseSender().send(errorMessageJson);
return true;
}
});
next.handleRequest(exchange);
}
El método implementado no tiene mayor complejidad, simplemente se extrae la información deseada de la excepción producida, se formatea en un objeto ErrorMessage y se devuelve la respuesta al usuario junto a un StatusCode 500.
Conclusiones
En conclusión, durante el presente artículo se ha tratado de detallar tanto buenas prácticas como algunos de los detalles de implementación que más complicados pueden resultar a la hora de trabajar con Undertow, como los handlers no bloqueantes o la lectura del body de la request y response, entre otros.
Referencias
Se recomienda encarecidamente leer los siguientes artículos que han servido de base para el escrito:
Mikel's TechBlog 