Executor Info Conductor Support (#32)

Author: devhawk

src/main/java/dev/dbos/transact/conductor/Conductor.java

@@ -12,6 +12,7 @@
 import dev.dbos.transact.workflow.WorkflowStatus;
 import dev.dbos.transact.workflow.internal.GetPendingWorkflowsOutput;
 
+import java.net.InetAddress;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.WebSocket;
@@ -341,9 +342,13 @@ BaseResponse getResponse(BaseMessage message) {
     }
 
     static BaseResponse handleExecutorInfo(Conductor conductor, BaseMessage message) {
-        // TODO: real implementation
-        ExecutorInfoRequest request = (ExecutorInfoRequest) message;
-        return new ExecutorInfoResponse(request, new RuntimeException("not yet implemented"));
+        try {
+            String hostname = InetAddress.getLocalHost().getHostName();
+            return new ExecutorInfoResponse(message, conductor.dbosExecutor.getExecutorId(),
+                    conductor.dbosExecutor.getAppVersion(), hostname);
+        } catch (Exception e) {
+            return new ExecutorInfoResponse(message, e);
+        }
     }
 
     static BaseResponse handleRecovery(Conductor conductor, BaseMessage message) {

src/main/java/dev/dbos/transact/conductor/protocol/ExecutorInfoRequest.java

@@ -2,4 +2,12 @@
 
 public class ExecutorInfoRequest extends BaseMessage {
     // empty on purpose
+
+    public ExecutorInfoRequest() {
+    }
+
+    public ExecutorInfoRequest(String requestId) {
+        this.type = MessageType.EXECUTOR_INFO.getValue();
+        this.request_id = requestId;
+    }
 }

src/main/java/dev/dbos/transact/conductor/protocol/ListQueuedWorkflowsRequest.java

@@ -87,11 +87,11 @@ public ListQueuedWorkflowsRequest build(String requestId) {
             body.start_time = this.start_time;
             body.end_time = this.end_time;
             body.status = this.status;
-            body.queue_name= this.queue_name;
-            body.limit= this.limit;
-            body.offset= this.offset;
-            body.sort_desc= this.sort_desc;
-            body.load_input= this.load_input;
+            body.queue_name = this.queue_name;
+            body.limit = this.limit;
+            body.offset = this.offset;
+            body.sort_desc = this.sort_desc;
+            body.load_input = this.load_input;
             request.body = body;
             return request;
         }

src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

@@ -59,6 +59,14 @@ public String getAppName() {
         return config.getName();
     }
 
+    public String getExecutorId() {
+        return GlobalParams.getInstance().getExecutorId();
+    }
+
+    public String getAppVersion() {
+        return GlobalParams.getInstance().getAppVersion();
+    }
+
     public void setQueueService(QueueService queueService) {
         this.queueService = queueService;
     }

src/test/java/dev/dbos/transact/conductor/ConductorTests.java

@@ -2,6 +2,7 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
@@ -10,6 +11,7 @@
 import dev.dbos.transact.conductor.TestWebSocketServer.WebSocketTestListener;
 import dev.dbos.transact.conductor.protocol.BaseMessage;
 import dev.dbos.transact.conductor.protocol.CancelRequest;
+import dev.dbos.transact.conductor.protocol.ExecutorInfoRequest;
 import dev.dbos.transact.conductor.protocol.ExistPendingWorkflowsRequest;
 import dev.dbos.transact.conductor.protocol.ForkWorkflowRequest;
 import dev.dbos.transact.conductor.protocol.GetWorkflowRequest;
@@ -29,6 +31,7 @@
 import dev.dbos.transact.workflow.WorkflowStatus;
 import dev.dbos.transact.workflow.internal.GetPendingWorkflowsOutput;
 
+import java.net.InetAddress;
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.List;
@@ -237,6 +240,36 @@ public void send(BaseMessage message) throws Exception {
         }
     }
 
+    @Test
+    public void canExecutorInfo() throws Exception {
+        MessageListener listener = new MessageListener();
+        testServer.setListener(listener);
+
+        String hostname = InetAddress.getLocalHost().getHostName();
+
+        when(mockExec.getAppVersion()).thenReturn("test-app-version");
+        when(mockExec.getExecutorId()).thenReturn("test-executor-id");
+
+        try (Conductor conductor = builder.build()) {
+            conductor.start();
+            assertTrue(listener.openLatch.await(5, TimeUnit.SECONDS), "open latch timed out");
+
+            ExecutorInfoRequest req = new ExecutorInfoRequest("12345");
+            listener.send(req);
+            assertTrue(listener.messageLatch.await(1000000000, TimeUnit.SECONDS), "message latch timed out");
+
+            JsonNode jsonNode = mapper.readTree(listener.message);
+            assertNotNull(jsonNode);
+            assertEquals("executor_info", jsonNode.get("type").asText());
+            assertEquals("12345", jsonNode.get("request_id").asText());
+            assertEquals(hostname, jsonNode.get("hostname").asText());
+            assertEquals("test-app-version", jsonNode.get("application_version").asText());
+            assertEquals("test-executor-id", jsonNode.get("executor_id").asText());
+            assertNull(jsonNode.get("error_message"));
+
+        }
+    }
+
     @Test
     public void canCancel() throws Exception {
         MessageListener listener = new MessageListener();
@@ -564,7 +597,8 @@ public void canListQueuedWorkflows() throws Exception {
                     .build("12345");
             listener.send(req);
             assertTrue(listener.messageLatch.await(100000000, TimeUnit.SECONDS), "message latch timed out");
-            ArgumentCaptor<ListQueuedWorkflowsInput> inputCaptor = ArgumentCaptor.forClass(ListQueuedWorkflowsInput.class);
+            ArgumentCaptor<ListQueuedWorkflowsInput> inputCaptor = ArgumentCaptor
+                    .forClass(ListQueuedWorkflowsInput.class);
             verify(mockDB).getQueuedWorkflows(inputCaptor.capture(), eq(false));
             ListQueuedWorkflowsInput input = inputCaptor.getValue();
             assertEquals(OffsetDateTime.parse("2024-06-01T12:34:56Z"), input.getStartTime());
@@ -585,7 +619,6 @@ public void canListQueuedWorkflows() throws Exception {
         }
     }
 
-
     @Test
     public void canGetWorkflow() throws Exception {
         MessageListener listener = new MessageListener();