Merge branch 'develop' into feature/27504/aws

Author: LEDfan

src/main/java/eu/openanalytics/containerproxy/api/ProxyRouteController.java

@@ -62,7 +62,7 @@ public void route(HttpServletRequest request, HttpServletResponse response) {
 			}
 
 			if (hasAccess) {
-				mappingManager.dispatchAsync(mapping, request, response);
+				mappingManager.dispatchAsync(proxyId, mapping, request, response);
 			} else {
 				response.setStatus(403);
 				response.getWriter().write("Not authorized to access this proxy");

src/main/java/eu/openanalytics/containerproxy/service/ProxyService.java

@@ -41,9 +41,8 @@
 import eu.openanalytics.containerproxy.spec.expression.SpecExpressionContext;
 import eu.openanalytics.containerproxy.spec.expression.SpecExpressionResolver;
 import eu.openanalytics.containerproxy.util.ProxyMappingManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.core.env.Environment;
 import org.springframework.data.util.Pair;
 import org.springframework.security.access.AccessDeniedException;
@@ -520,7 +519,6 @@ private boolean cleanupIfPendingAppWasStopped(Proxy startingProxy) {
 		return false;
 	}
 
-
 	/**
 	 * Add existing Proxy to the ProxyService.
 	 * This is used by the AppRecovery feature.

src/main/java/eu/openanalytics/containerproxy/util/ProxyMappingManager.java

@@ -20,7 +20,14 @@
  */
 package eu.openanalytics.containerproxy.util;
 
+import eu.openanalytics.containerproxy.model.runtime.Proxy;
+import eu.openanalytics.containerproxy.model.runtime.ProxyStatus;
+import eu.openanalytics.containerproxy.service.AsyncProxyService;
+import eu.openanalytics.containerproxy.service.ProxyService;
+import eu.openanalytics.containerproxy.service.StructuredLogger;
 import eu.openanalytics.containerproxy.service.hearbeat.HeartbeatService;
+import io.undertow.io.Sender;
+import io.undertow.server.DefaultResponseListener;
 import io.undertow.server.HttpHandler;
 import io.undertow.server.HttpServerExchange;
 import io.undertow.server.handlers.PathHandler;
@@ -31,8 +38,10 @@
 import io.undertow.server.handlers.proxy.ProxyHandler;
 import io.undertow.servlet.handlers.ServletRequestContext;
 import io.undertow.util.AttachmentKey;
+import io.undertow.util.Headers;
 import io.undertow.util.PathMatcher;
 import org.springframework.context.annotation.Lazy;
+import io.undertow.util.StatusCodes;
 import org.springframework.stereotype.Component;
 
 import javax.inject.Inject;
@@ -58,15 +67,26 @@ public class ProxyMappingManager {
 
 	private static final String PROXY_INTERNAL_ENDPOINT = "/proxy_endpoint";
 	private static final AttachmentKey<ProxyMappingManager> ATTACHMENT_KEY_DISPATCHER = AttachmentKey.create(ProxyMappingManager.class);
-	
+	private static final AttachmentKey<ProxyIdAttachment> ATTACHMENT_KEY_PROXY_ID = AttachmentKey.create(ProxyIdAttachment.class);
+
+	private final StructuredLogger logger = StructuredLogger.create(getClass());
+
 	private PathHandler pathHandler;
 
 	private final Map<String, String> mappings = new HashMap<>();
 
 	@Inject
 	@Lazy
 	private HeartbeatService heartbeatService;
-	
+
+	@Inject
+	@Lazy
+	private ProxyService proxyService;
+
+	@Inject
+	@Lazy
+	private AsyncProxyService asyncProxyService;
+
 	public synchronized HttpHandler createHttpHandler(HttpHandler defaultHandler) {
 		if (pathHandler == null) {
 			pathHandler = new ProxyPathHandler(defaultHandler);
@@ -123,32 +143,70 @@ public String getProxyId(String mapping) {
 	 * 
 	 * Note that clients can never access a proxy handler directly (for security reasons).
 	 * Dispatching is the only allowed method to access proxy handlers.
-	 * 
+	 *
+	 * @param proxyId The id of the proxy
 	 * @param mapping The target mapping to dispatch to.
 	 * @param request The request to dispatch.
 	 * @param response The response corresponding to the request.
 	 * @throws IOException If the dispatch fails for an I/O reason.
 	 * @throws ServletException If the dispatch fails for any other reason.
 	 */
-	public void dispatchAsync(String mapping, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
-		dispatchAsync(mapping, request, response, null);
+	public void dispatchAsync(String proxyId, String mapping, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
+		dispatchAsync(proxyId, mapping, request, response, null);
 	}
 
-	public void dispatchAsync(String mapping, HttpServletRequest request, HttpServletResponse response, Consumer<HttpServerExchange> exchangeCustomizer) throws IOException, ServletException {
+	public void dispatchAsync(String proxyId, String mapping, HttpServletRequest request, HttpServletResponse response, Consumer<HttpServerExchange> exchangeCustomizer) throws IOException, ServletException {
 		HttpServerExchange exchange = ServletRequestContext.current().getExchange();
 		exchange.putAttachment(ATTACHMENT_KEY_DISPATCHER, this);
-
+		exchange.putAttachment(ATTACHMENT_KEY_PROXY_ID, new ProxyIdAttachment(proxyId));
+		
 		String queryString = request.getQueryString();
 		queryString = (queryString == null) ? "" : "?" + queryString;
 		String targetPath = PROXY_INTERNAL_ENDPOINT + "/" + mapping + queryString;
 
 		if (exchangeCustomizer != null) {
 			exchangeCustomizer.accept(exchange);
 		}
+		exchange.addDefaultResponseListener(defaultResponseListener);
 		request.startAsync();
 		request.getRequestDispatcher(targetPath).forward(request, response);
 	}
 
+	private final DefaultResponseListener defaultResponseListener = responseExchange -> {
+		if (!responseExchange.isResponseChannelAvailable()) {
+			return false;
+		}
+		if (responseExchange.getStatusCode() == StatusCodes.SERVICE_UNAVAILABLE) {
+			ProxyIdAttachment proxyIdAttachment = responseExchange.getAttachment(ATTACHMENT_KEY_PROXY_ID);
+			Proxy proxy = null;
+			if (proxyIdAttachment != null) {
+				try {
+					proxy = proxyService.getProxy(proxyIdAttachment.proxyId);
+					if (proxy != null) {
+						logger.info(proxy, "Proxy unreachable/crashed, stopping it now");
+						asyncProxyService.stopProxy(proxy, true);
+					}
+				} catch (Throwable t) {
+					// ignore in order to complete request
+				}
+			}
+
+			String errorPage;
+			if (proxy != null && proxy.getStatus() != ProxyStatus.Stopped) {
+				errorPage = "{\"status\":\"error\", \"message\":\"app_crashed\"}";
+			} else {
+				// in-progress request got terminated because the app has been stopped (not crashed)
+				errorPage = "{\"status\":\"error\", \"message\":\"app_stopped_or_non_existent\"}";
+			}
+			responseExchange.getResponseHeaders().put(Headers.CONTENT_LENGTH, "" + errorPage.length());
+			responseExchange.getResponseHeaders().put(Headers.CONTENT_TYPE, "application/json");
+			Sender sender = responseExchange.getResponseSender();
+			sender.send(errorPage);
+			return true;
+		}
+		return false;
+	};
+	
 	private static class ProxyPathHandler extends PathHandler {
 
 		public ProxyPathHandler(HttpHandler defaultHandler) {
@@ -173,4 +231,13 @@ public void handleRequest(HttpServerExchange exchange) throws Exception {
 			}
 		}
 	}
+
+	private static class ProxyIdAttachment {
+		final String proxyId;
+
+		public ProxyIdAttachment(String proxyId) {
+			this.proxyId = proxyId;
+		}
+	}
+
 }