Support sockets as transport for framed telemetry (aws#378)

luben · web-flow · commit 04ef611147fb · 2022-11-15T16:35:23.000Z
Sockets cannot be re-opened (unlike pipes and regular files). We have
to reflectively construct the `FileDescriptor` with the right number as
it does not have a constructor that takes int.
diff --git a/aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/AWSLambda.java b/aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/AWSLambda.java
@@ -24,10 +24,12 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.FileDescriptor;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.PrintStream;
+import java.lang.reflect.Constructor;
 import java.lang.reflect.Method;
 import java.net.URLClassLoader;
 import java.nio.file.Paths;
@@ -169,16 +171,31 @@ public static String getEnvOrExit(String envVariableName) {
 
     protected static URLClassLoader customerClassLoader;
 
+    /**
+     * convert an integer into a FileDescriptor object using reflection to access private members.
+     */
+    private static FileDescriptor intToFd(int fd) throws RuntimeException {
+        try {
+            Class<FileDescriptor> clazz = FileDescriptor.class;
+            Constructor<FileDescriptor> c = clazz.getDeclaredConstructor(new Class<?>[] { Integer.TYPE });
+            c.setAccessible(true);
+            return c.newInstance(new Integer(fd));
+        } catch(Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     private static LogSink createLogSink() {
-        final String fd = System.getenv("_LAMBDA_TELEMETRY_LOG_FD");
-        if(fd == null) {
+        final String fdStr = System.getenv("_LAMBDA_TELEMETRY_LOG_FD");
+        if(fdStr == null) {
             return new StdOutLogSink();
         }
 
         try {
-            File pipeFdFile = Paths.get("/proc", "self", "fd", fd).toFile();
-            return new FramedTelemetryLogSink(pipeFdFile);
-        } catch (IOException e) {
+            int fdInt = Integer.parseInt(fdStr);
+            FileDescriptor fd = intToFd(fdInt);
+            return new FramedTelemetryLogSink(fd);
+        } catch (Exception e) {
             return new StdOutLogSink();
         }
     }
diff --git a/aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/logging/FramedTelemetryLogSink.java b/aws-lambda-java-runtime-interface-client/src/main/java/com/amazonaws/services/lambda/runtime/api/client/logging/FramedTelemetryLogSink.java
@@ -2,7 +2,7 @@
 
 package com.amazonaws.services.lambda.runtime.api.client.logging;
 
-import java.io.File;
+import java.io.FileDescriptor;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -31,8 +31,8 @@ public class FramedTelemetryLogSink implements LogSink {
     private final FileOutputStream logOutputStream;
     private final ByteBuffer headerBuf;
 
-    public FramedTelemetryLogSink(File file) throws IOException {
-        this.logOutputStream = new FileOutputStream(file);
+    public FramedTelemetryLogSink(FileDescriptor fd) throws IOException {
+        this.logOutputStream = new FileOutputStream(fd);
         this.headerBuf = ByteBuffer.allocate(HEADER_LENGTH).order(ByteOrder.BIG_ENDIAN);
     }
 
diff --git a/aws-lambda-java-runtime-interface-client/src/test/java/com/amazonaws/services/lambda/runtime/api/client/logging/FramedTelemetryLogSinkTest.java b/aws-lambda-java-runtime-interface-client/src/test/java/com/amazonaws/services/lambda/runtime/api/client/logging/FramedTelemetryLogSinkTest.java
@@ -6,7 +6,9 @@
 import org.junit.jupiter.api.io.TempDir;
 
 import java.io.File;
+import java.io.FileDescriptor;
 import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
@@ -29,7 +31,9 @@ public class FramedTelemetryLogSinkTest {
     public void logSingleFrame() throws IOException {
         byte[] message = "hello world\nsomething on a new line!\n".getBytes();
         File tmpFile = tmpFolder.resolve("pipe").toFile();
-        try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(tmpFile)) {
+        FileOutputStream fos = new FileOutputStream(tmpFile);
+        FileDescriptor fd = fos.getFD();
+        try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(fd)) {
             logSink.log(message);
         }
 
@@ -63,7 +67,9 @@ public void logMultipleFrames() throws IOException {
         byte[] firstMessage = "hello world\nsomething on a new line!".getBytes();
         byte[] secondMessage = "hello again\nhere's another message\n".getBytes();
         File tmpFile = tmpFolder.resolve("pipe").toFile();
-        try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(tmpFile)) {
+        FileOutputStream fos = new FileOutputStream(tmpFile);
+        FileDescriptor fd = fos.getFD();
+        try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(fd)) {
             logSink.log(firstMessage);
             logSink.log(secondMessage);
         }
@@ -107,7 +113,9 @@ public void interruptedThread() throws IOException {
         try {
             byte[] message = "hello world\nsomething on a new line!\n".getBytes();
             File tmpFile = tmpFolder.resolve("pipe").toFile();
-            try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(tmpFile)) {
+            FileOutputStream fos = new FileOutputStream(tmpFile);
+            FileDescriptor fd = fos.getFD();
+            try (FramedTelemetryLogSink logSink = new FramedTelemetryLogSink(fd)) {
                 Thread.currentThread().interrupt();
 
                 logSink.log(message);
