Add support for PROXY protocol

core/src/main/java/io/confluent/rest/ApplicationServer.java

@@ -27,10 +27,12 @@
 import org.eclipse.jetty.http2.server.HTTP2ServerConnectionFactory;
 import org.eclipse.jetty.http2.server.HTTP2CServerConnectionFactory;
 import org.eclipse.jetty.jmx.MBeanContainer;
+import org.eclipse.jetty.server.ConnectionFactory;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.HttpConnectionFactory;
 import org.eclipse.jetty.server.NetworkTrafficServerConnector;
+import org.eclipse.jetty.server.ProxyConnectionFactory;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.SslConnectionFactory;
 import org.eclipse.jetty.server.handler.ContextHandlerCollection;
@@ -416,6 +418,9 @@ private void configureConnectors(SslContextFactory sslContextFactory) {
     final boolean http2Enabled = isJava11Compatible()
                               && config.getBoolean(RestConfig.HTTP2_ENABLED_CONFIG);
 
+    final boolean proxyProtocolEnabled =
+        config.getBoolean(RestConfig.PROXY_PROTOCOL_ENABLED_CONFIG);
+
     @SuppressWarnings("deprecation")
     List<NamedURI> listeners = parseListeners(
         config.getList(RestConfig.LISTENERS_CONFIG),
@@ -424,69 +429,104 @@ private void configureConnectors(SslContextFactory sslContextFactory) {
         SUPPORTED_URI_SCHEMES, "http");
 
     for (NamedURI listener : listeners) {
-      addConnectorForListener(httpConfiguration, httpConnectionFactory, listener, http2Enabled);
+      addConnectorForListener(httpConfiguration, httpConnectionFactory, listener,
+              http2Enabled, proxyProtocolEnabled);
     }
   }
 
   private void addConnectorForListener(HttpConfiguration httpConfiguration,
                                        HttpConnectionFactory httpConnectionFactory,
                                        NamedURI listener,
-                                       boolean http2Enabled) {
-    NetworkTrafficServerConnector connector;
+                                       boolean http2Enabled,
+                                       boolean proxyProtocolEnabled) {
+    ConnectionFactory[] connectionFactories = getConnectionFactories(httpConfiguration,
+        httpConnectionFactory, listener, http2Enabled, proxyProtocolEnabled);
+    NetworkTrafficServerConnector connector = new NetworkTrafficServerConnector(this, null, null,
+        null, 0, 0, connectionFactories);
+    if (http2Enabled) {
+      // In Jetty 9.4.37, there was a change in behaviour to implement RFC 7230 more
+      // rigorously and remove support for ambiguous URIs, such as escaping
+      // . and / characters. While this is a good idea because it prevents clever
+      // path-manipulation tricks in URIs, it breaks certain existing systems including
+      // the Schema Registry. Jetty behaviour was then reverted specifying the compliance mode
+      // in the HttpConnectionFactory class using the HttpCompliance.RFC7230 enum. This has
+      // the problem that it applies only to HTTP/1.1. The following sets this compliance mode
+      // explicitly when HTTP/2 is enabled.
+      connector.addBean(HttpCompliance.RFC7230);
+    }
+
+    connector.setPort(listener.getUri().getPort());
+    connector.setHost(listener.getUri().getHost());
+    connector.setIdleTimeout(config.getLong(RestConfig.IDLE_TIMEOUT_MS_CONFIG));
+    if (listener.getName() != null) {
+      connector.setName(listener.getName());
+    }
+
+    connectors.add(connector);
+    super.addConnector(connector);
+  }
+
+  private ConnectionFactory[] getConnectionFactories(HttpConfiguration httpConfiguration,
+                                                     HttpConnectionFactory httpConnectionFactory,
+                                                     NamedURI listener,
+                                                     boolean http2Enabled,
+                                                     boolean proxyProtocolEnabled) {
+    ArrayList<ConnectionFactory> connectionFactories = new ArrayList<>();
 
     if (http2Enabled) {
       log.info("Adding listener with HTTP/2: " + listener);
       if (listener.getUri().getScheme().equals("http")) {
         // HTTP2C is HTTP/2 Clear text
         final HTTP2CServerConnectionFactory h2cConnectionFactory =
-                new HTTP2CServerConnectionFactory(httpConfiguration);
+            new HTTP2CServerConnectionFactory(httpConfiguration);
+
+        if (proxyProtocolEnabled) {
+          connectionFactories.add(new ProxyConnectionFactory(httpConnectionFactory.getProtocol()));
+        }
 
         // The order of HTTP and HTTP/2 is significant here but it's not clear why :)
-        connector = new NetworkTrafficServerConnector(this, null, null, null, 0, 0,
-                httpConnectionFactory, h2cConnectionFactory);
+        connectionFactories.add(httpConnectionFactory);
+        connectionFactories.add(h2cConnectionFactory);
       } else {
         final HTTP2ServerConnectionFactory h2ConnectionFactory =
-                new HTTP2ServerConnectionFactory(httpConfiguration);
+            new HTTP2ServerConnectionFactory(httpConfiguration);
 
         ALPNServerConnectionFactory alpnConnectionFactory = new ALPNServerConnectionFactory();
         alpnConnectionFactory.setDefaultProtocol(HttpVersion.HTTP_1_1.asString());
 
         SslConnectionFactory sslConnectionFactory = new SslConnectionFactory(sslContextFactory,
-                alpnConnectionFactory.getProtocol());
+            alpnConnectionFactory.getProtocol());
 
-        connector = new NetworkTrafficServerConnector(this, null, null, null, 0, 0,
-                sslConnectionFactory, alpnConnectionFactory, h2ConnectionFactory,
-                httpConnectionFactory);
-      }
+        if (proxyProtocolEnabled) {
+          connectionFactories.add(new ProxyConnectionFactory(sslConnectionFactory.getProtocol()));
+        }
 
-      // In Jetty 9.4.37, there was a change in behaviour to implement RFC 7230 more
-      // rigorously and remove support for ambiguous URIs, such as escaping
-      // . and / characters. While this is a good idea because it prevents clever
-      // path-manipulation tricks in URIs, it breaks certain existing systems including
-      // the Schema Registry. Jetty behaviour was then reverted specifying the compliance mode
-      // in the HttpConnectionFactory class using the HttpCompliance.RFC7230 enum. This has
-      // the problem that it applies only to HTTP/1.1. The following sets this compliance mode
-      // explicitly when HTTP/2 is enabled.
-      connector.addBean(HttpCompliance.RFC7230);
+        connectionFactories.add(sslConnectionFactory);
+        connectionFactories.add(alpnConnectionFactory);
+        connectionFactories.add(h2ConnectionFactory);
+        connectionFactories.add(httpConnectionFactory);
+      }
     } else {
       log.info("Adding listener: " + listener);
       if (listener.uri.getScheme().equals("http")) {
-        connector = new NetworkTrafficServerConnector(this, httpConnectionFactory);
+        if (proxyProtocolEnabled) {
+          connectionFactories.add(new ProxyConnectionFactory(httpConnectionFactory.getProtocol()));
+        }
+
       } else {
-        connector = new NetworkTrafficServerConnector(this, httpConnectionFactory,
-                sslContextFactory);
-      }
-    }
+        SslConnectionFactory sslConnectionFactory = new SslConnectionFactory(sslContextFactory,
+            httpConnectionFactory.getProtocol());
 
-    connector.setPort(listener.getUri().getPort());
-    connector.setHost(listener.getUri().getHost());
-    connector.setIdleTimeout(config.getLong(RestConfig.IDLE_TIMEOUT_MS_CONFIG));
-    if (listener.getName() != null) {
-      connector.setName(listener.getName());
+        if (proxyProtocolEnabled) {
+          connectionFactories.add(new ProxyConnectionFactory(sslConnectionFactory.getProtocol()));
+        }
+
+        connectionFactories.add(sslConnectionFactory);
+      }
+      connectionFactories.add(httpConnectionFactory);
     }
 
-    connectors.add(connector);
-    super.addConnector(connector);
+    return connectionFactories.toArray(new ConnectionFactory[0]);
   }
 
   /**

core/src/main/java/io/confluent/rest/RestConfig.java

@@ -446,6 +446,20 @@ public class RestConfig extends AbstractConfig {
           + "Java 11 JVM or later. Default is true.";
   protected static final boolean HTTP2_ENABLED_DEFAULT = true;
 
+  public static final String PROXY_PROTOCOL_ENABLED_CONFIG =
+      "proxy.protocol.enabled";
+  protected static final String PROXY_PROTOCOL_ENABLED_DOC =
+      "If true, enable support for the PROXY protocol. When enabled, the server will "
+          + "automatically detect whether PROXY protocol headers are present, and if they are, "
+          + "whether V1 or V2 of the protocol is in use. When the headers are present, the actual "
+          + "IP address and port of the client will be visible when the request is handled. If "
+          + "the headers are not present, requests will be processed normally, but if the server "
+          + "runs behind a load balancer, the request will appear to come from the the IP address "
+          + "and port of the load balancer. See "
+          + "https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt for more information. "
+          + "Default is false.";
+  protected static final boolean PROXY_PROTOCOL_ENABLED_DEFAULT = false;
+
   public static ConfigDef baseConfigDef() {
     return baseConfigDef(
         PORT_CONFIG_DEFAULT,
@@ -921,6 +935,12 @@ private static ConfigDef incompleteBaseConfigDef() {
             LISTENER_PROTOCOL_MAP_DEFAULT,
             Importance.LOW,
             LISTENER_PROTOCOL_MAP_DOC
+        ).define(
+            PROXY_PROTOCOL_ENABLED_CONFIG,
+            Type.BOOLEAN,
+            PROXY_PROTOCOL_ENABLED_DEFAULT,
+            Importance.LOW,
+            PROXY_PROTOCOL_ENABLED_DOC
         );
   }