Implement HTTP/1.1 access logging

Author: Azeem Muzammil

ballerina/Ballerina.toml

@@ -1,7 +1,7 @@
 [package]
 org = "ballerina"
 name = "http"
-version = "2.10.13"
+version = "2.10.14"
 authors = ["Ballerina"]
 keywords = ["http", "network", "service", "listener", "client"]
 repository = "https://github.com/ballerina-platform/module-ballerina-http"
@@ -16,8 +16,8 @@ graalvmCompatible = true
 [[platform.java17.dependency]]
 groupId = "io.ballerina.stdlib"
 artifactId = "http-native"
-version = "2.10.13"
-path = "../native/build/libs/http-native-2.10.13.jar"
+version = "2.10.14"
+path = "../native/build/libs/http-native-2.10.14-SNAPSHOT.jar"
 
 [[platform.java17.dependency]]
 groupId = "io.ballerina.stdlib"

ballerina/CompilerPlugin.toml

@@ -3,4 +3,4 @@ id = "http-compiler-plugin"
 class = "io.ballerina.stdlib.http.compiler.HttpCompilerPlugin"
 
 [[dependency]]
-path = "../compiler-plugin/build/libs/http-compiler-plugin-2.10.13.jar"
+path = "../compiler-plugin/build/libs/http-compiler-plugin-2.10.14-SNAPSHOT.jar"

ballerina/Dependencies.toml

@@ -25,7 +25,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "cache"
-version = "3.7.0"
+version = "3.7.1"
 dependencies = [
 	{org = "ballerina", name = "constraint"},
 	{org = "ballerina", name = "jballerina.java"},
@@ -76,7 +76,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "http"
-version = "2.10.13"
+version = "2.10.14"
 dependencies = [
 	{org = "ballerina", name = "auth"},
 	{org = "ballerina", name = "cache"},
@@ -283,7 +283,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "observe"
-version = "1.2.0"
+version = "1.2.3"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"}
 ]

native/src/main/java/io/ballerina/stdlib/http/transport/contractimpl/DefaultHttpClientConnector.java

@@ -198,7 +198,6 @@ public HttpResponseFuture send(OutboundMsgHolder outboundMsgHolder, HttpCarbonMe
                     new RequestWriteStarter(outboundMsgHolder, activeHttp2ClientChannel).startWritingContent();
                     httpResponseFuture = outboundMsgHolder.getResponseFuture();
                     httpResponseFuture.notifyResponseHandle(new ResponseHandle(outboundMsgHolder));
-
                     return httpResponseFuture;
                 }
             }

native/src/main/java/io/ballerina/stdlib/http/transport/contractimpl/HttpOutboundRespListener.java

@@ -35,6 +35,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Calendar;
 import java.util.Locale;
 
 /**
@@ -53,6 +54,8 @@ public class HttpOutboundRespListener implements HttpConnectorListener {
     private ChunkConfig chunkConfig;
     private KeepAliveConfig keepAliveConfig;
     private String serverName;
+    private Calendar inboundRequestArrivalTime;
+    private String remoteAddress = "-";
 
     public HttpOutboundRespListener(HttpCarbonMessage requestMsg, SourceHandler sourceHandler) {
         this.requestDataHolder = new RequestDataHolder(requestMsg);
@@ -64,6 +67,8 @@ public HttpOutboundRespListener(HttpCarbonMessage requestMsg, SourceHandler sour
         this.handlerExecutor = HttpTransportContextHolder.getInstance().getHandlerExecutor();
         this.serverName = sourceHandler.getServerName();
         this.listenerReqRespStateManager = requestMsg.listenerReqRespStateManager;
+        this.remoteAddress = sourceHandler.getRemoteHost();
+        this.inboundRequestArrivalTime = Calendar.getInstance();
         setBackPressureObservableToHttpResponseFuture();
     }
 
@@ -146,4 +151,12 @@ public void setKeepAliveConfig(KeepAliveConfig config) {
     public SourceHandler getSourceHandler() {
         return sourceHandler;
     }
+
+    public Calendar getInboundRequestArrivalTime() {
+        return inboundRequestArrivalTime;
+    }
+
+    public String getRemoteAddress() {
+        return remoteAddress;
+    }
 }

native/src/main/java/io/ballerina/stdlib/http/transport/contractimpl/listener/HttpServerChannelInitializer.java

@@ -66,8 +66,6 @@
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
 
-import static io.ballerina.stdlib.http.transport.contract.Constants.ACCESS_LOG;
-import static io.ballerina.stdlib.http.transport.contract.Constants.HTTP_ACCESS_LOG_HANDLER;
 import static io.ballerina.stdlib.http.transport.contract.Constants.HTTP_TRACE_LOG_HANDLER;
 import static io.ballerina.stdlib.http.transport.contract.Constants.MAX_ENTITY_BODY_VALIDATION_HANDLER;
 import static io.ballerina.stdlib.http.transport.contract.Constants.SECURITY;
@@ -220,9 +218,9 @@ public void configureHttpPipeline(ChannelPipeline serverPipeline, String initial
             if (httpTraceLogEnabled) {
                 serverPipeline.addLast(HTTP_TRACE_LOG_HANDLER, new HttpTraceLoggingHandler(TRACE_LOG_DOWNSTREAM));
             }
-            if (httpAccessLogEnabled) {
-                serverPipeline.addLast(HTTP_ACCESS_LOG_HANDLER, new HttpAccessLoggingHandler(ACCESS_LOG));
-            }
+//            if (httpAccessLogEnabled) {
+//                serverPipeline.addLast(HTTP_ACCESS_LOG_HANDLER, new HttpAccessLoggingHandler(ACCESS_LOG));
+//            }
         }
         serverPipeline.addLast(URI_HEADER_LENGTH_VALIDATION_HANDLER, new UriAndHeaderLengthValidator(this.serverName));
         if (reqSizeValidationConfig.getMaxEntityBodySize() > -1) {
@@ -235,7 +233,7 @@ public void configureHttpPipeline(ChannelPipeline serverPipeline, String initial
                                                                    webSocketCompressionEnabled));
         serverPipeline.addLast(Constants.BACK_PRESSURE_HANDLER, new BackPressureHandler());
         serverPipeline.addLast(Constants.HTTP_SOURCE_HANDLER,
-                               new SourceHandler(this.serverConnectorFuture, this.interfaceId, this.chunkConfig,
+                               new SourceHandler(this.serverConnectorFuture, this, this.interfaceId, this.chunkConfig,
                                                  keepAliveConfig, this.serverName, this.allChannels,
                                                  this.listenerChannels, this.pipeliningEnabled, this.pipeliningLimit,
                                                  this.pipeliningGroup));
@@ -267,9 +265,9 @@ private void configureH2cPipeline(ChannelPipeline pipeline) {
             pipeline.addLast(HTTP_TRACE_LOG_HANDLER,
                              new HttpTraceLoggingHandler(TRACE_LOG_DOWNSTREAM));
         }
-        if (httpAccessLogEnabled) {
-            pipeline.addLast(HTTP_ACCESS_LOG_HANDLER, new HttpAccessLoggingHandler(ACCESS_LOG));
-        }
+//        if (httpAccessLogEnabled) {
+//            pipeline.addLast(HTTP_ACCESS_LOG_HANDLER, new HttpAccessLoggingHandler(ACCESS_LOG));
+//        }
 
         final HttpServerUpgradeHandler.UpgradeCodecFactory upgradeCodecFactory = protocol -> {
             if (AsciiString.contentEquals(Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME, protocol)) {

native/src/main/java/io/ballerina/stdlib/http/transport/contractimpl/listener/SourceHandler.java

@@ -19,6 +19,7 @@
 
 package io.ballerina.stdlib.http.transport.contractimpl.listener;
 
+import io.ballerina.stdlib.http.api.logging.accesslog.HttpAccessLogMessage;
 import io.ballerina.stdlib.http.transport.contract.Constants;
 import io.ballerina.stdlib.http.transport.contract.ServerConnectorFuture;
 import io.ballerina.stdlib.http.transport.contract.config.ChunkConfig;
@@ -46,7 +47,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.PriorityQueue;
 import java.util.Queue;
@@ -73,8 +77,10 @@ public class SourceHandler extends ChannelInboundHandlerAdapter {
 
     private KeepAliveConfig keepAliveConfig;
     private ServerConnectorFuture serverConnectorFuture;
+    private HttpServerChannelInitializer serverChannelInitializer;
     private String interfaceId;
     private String serverName;
+    private String remoteHost;
     private boolean idleTimeout;
     private ChannelGroup allChannels;
     private ChannelGroup listenerChannels;
@@ -87,12 +93,15 @@ public class SourceHandler extends ChannelInboundHandlerAdapter {
     private long sequenceId = 1L; //Keep track of the request order for http 1.1 pipelining
     private final Queue holdingQueue = new PriorityQueue<>(NUMBER_OF_INITIAL_EVENTS_HELD);
     private EventExecutorGroup pipeliningGroup;
+    private List<HttpAccessLogMessage> httpAccessLogMessages;
 
-    public SourceHandler(ServerConnectorFuture serverConnectorFuture, String interfaceId, ChunkConfig chunkConfig,
-                         KeepAliveConfig keepAliveConfig, String serverName, ChannelGroup allChannels,
-                         ChannelGroup listenerChannels, boolean pipeliningEnabled, long pipeliningLimit,
-                         EventExecutorGroup pipeliningGroup) {
+    public SourceHandler(ServerConnectorFuture serverConnectorFuture,
+                         HttpServerChannelInitializer serverChannelInitializer, String interfaceId,
+                         ChunkConfig chunkConfig, KeepAliveConfig keepAliveConfig, String serverName,
+                         ChannelGroup allChannels, ChannelGroup listenerChannels, boolean pipeliningEnabled,
+                         long pipeliningLimit, EventExecutorGroup pipeliningGroup) {
         this.serverConnectorFuture = serverConnectorFuture;
+        this.serverChannelInitializer = serverChannelInitializer;
         this.interfaceId = interfaceId;
         this.chunkConfig = chunkConfig;
         this.keepAliveConfig = keepAliveConfig;
@@ -104,6 +113,7 @@ public SourceHandler(ServerConnectorFuture serverConnectorFuture, String interfa
         this.pipeliningEnabled = pipeliningEnabled;
         this.pipeliningLimit = pipeliningLimit;
         this.pipeliningGroup = pipeliningGroup;
+        this.httpAccessLogMessages = new ArrayList<>();
     }
 
     @SuppressWarnings("unchecked")
@@ -156,6 +166,12 @@ public void channelActive(final ChannelHandlerContext ctx) {
             handlerExecutor.executeAtSourceConnectionInitiation(Integer.toString(ctx.hashCode()));
         }
         this.remoteAddress = ctx.channel().remoteAddress();
+        if (this.remoteAddress instanceof InetSocketAddress) {
+            remoteHost = ((InetSocketAddress) this.remoteAddress).getAddress().toString();
+            if (remoteHost.startsWith("/")) {
+                remoteHost = remoteHost.substring(1);
+            }
+        }
     }
 
     @Override
@@ -317,6 +333,10 @@ public ServerConnectorFuture getServerConnectorFuture() {
         return serverConnectorFuture;
     }
 
+    public HttpServerChannelInitializer getServerChannelInitializer() {
+        return serverChannelInitializer;
+    }
+
     public ChunkConfig getChunkConfig() {
         return chunkConfig;
     }
@@ -329,6 +349,10 @@ public String getServerName() {
         return serverName;
     }
 
+    public String getRemoteHost() {
+        return remoteHost;
+    }
+
     public void setConnectedState(boolean connectedState) {
         this.connectedState = connectedState;
     }
@@ -340,4 +364,12 @@ public void removeRequestEntry(HttpCarbonMessage inboundRequestMsg) {
     public void resetInboundRequestMsg() {
         this.inboundRequestMsg = null;
     }
+
+    public void addHttpAccessLogMessage(HttpAccessLogMessage httpAccessLogMessage) {
+        this.httpAccessLogMessages.add(httpAccessLogMessage);
+    }
+
+    public List<HttpAccessLogMessage> getHttpAccessLogMessages() {
+        return this.httpAccessLogMessages;
+    }
 }