feat: stream manager debugging support

pom.xml

@@ -80,11 +80,31 @@
             <version>5.5.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.amazonaws.greengrass</groupId>
+            <artifactId>streammanager</artifactId>
+            <version>1.1.0</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.dataformat</groupId>
-            <artifactId>jackson-dataformat-yaml</artifactId>
-            <version>2.10.1</version>
-            <scope>provided</scope>
+            <artifactId>jackson-dataformat-cbor</artifactId>
+            <version>2.13.3</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-core</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.fasterxml.jackson.core</groupId>
+                    <artifactId>jackson-databind</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.mockito</groupId>

src/main/java/com/aws/greengrass/localdebugconsole/APICalls.java

@@ -96,4 +96,39 @@ public enum APICalls {
      * Utility called by the client to unsubscribe to a local IPC topic.
      */
     unsubscribeToPubSubTopic,
+
+    /**
+     * Returns the current Stream Manager streams list
+     */
+    streamManagerListStreams,
+
+    /**
+     * Describes a message stream to get metadata including the stream’s definition, size, and exporter statuses.
+     */
+    streamManagerDescribeStream,
+
+    /**
+     * Deletes a message stream based on its name.
+     */
+    streamManagerDeleteMessageStream,
+
+    /**
+     * Read message(s) from a chosen stream with options. If no options are specified it will try to read 1 message from the stream.
+     */
+    streamManagerReadMessages,
+
+    /**
+     * Append a message into the specified message stream.
+     */
+    streamManagerAppendMessage,
+
+    /**
+     * Create a message stream with a given definition.
+     */
+    streamManagerCreateMessageStream,
+
+    /**
+     * Update a message stream with a given definition.
+     */
+    streamManagerUpdateMessageStream,
 }

src/main/java/com/aws/greengrass/localdebugconsole/DashboardServer.java

@@ -5,6 +5,7 @@
 
 package com.aws.greengrass.localdebugconsole;
 
+import com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition;
 import com.aws.greengrass.builtin.services.pubsub.PubSubIPCEventStreamAgent;
 import com.aws.greengrass.builtin.services.pubsub.PublishEvent;
 import com.aws.greengrass.builtin.services.pubsub.SubscribeRequest;
@@ -16,6 +17,7 @@
 import com.aws.greengrass.localdebugconsole.messageutils.MessageType;
 import com.aws.greengrass.localdebugconsole.messageutils.PackedRequest;
 import com.aws.greengrass.localdebugconsole.messageutils.Request;
+import com.aws.greengrass.localdebugconsole.messageutils.StreamManagerResponseMessage;
 import com.aws.greengrass.logging.api.Logger;
 import com.aws.greengrass.mqttclient.MqttClient;
 import com.aws.greengrass.mqttclient.MqttRequestException;
@@ -24,6 +26,7 @@
 import com.aws.greengrass.mqttclient.v5.Unsubscribe;
 import com.aws.greengrass.util.DefaultConcurrentHashMap;
 import com.aws.greengrass.util.Pair;
+import com.aws.greengrass.util.Utils;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -71,20 +74,23 @@ public class DashboardServer extends WebSocketServer implements KernelMessagePus
     private final Authenticator authenticator;
     private final MqttClient mqttClient;
 
+    private final StreamManagerHelper streamManagerHelper;
+
     PubSubIPCEventStreamAgent pubSubIPCAgent;
     private final String SERVICE_NAME = "LocalDebugConsole";
 
     public DashboardServer(InetSocketAddress address, Logger logger, Kernel root, DeviceConfiguration deviceConfig,
-                           Authenticator authenticator, Provider<SSLEngine> engineProvider) {
+                           Authenticator authenticator, Provider<SSLEngine> engineProvider, String streamManagerAuthToken) {
         this(address, logger, new KernelCommunicator(root, logger, deviceConfig), authenticator, engineProvider,
                 root.getContext().get(PubSubIPCEventStreamAgent.class),
-                root.getContext().get(MqttClient.class));
+                root.getContext().get(MqttClient.class),
+                 new StreamManagerHelper(root, streamManagerAuthToken));
     }
 
     // constructor for unit testing
     DashboardServer(InetSocketAddress address, Logger logger, DashboardAPI dashboardAPI, Authenticator authenticator,
                     Provider<SSLEngine> engineProvider, PubSubIPCEventStreamAgent pubSubIPCAgent,
-                    MqttClient mqttClient) {
+                    MqttClient mqttClient, StreamManagerHelper streamManagerHelper) {
         super(address);
         setReuseAddr(true);
         setTcpNoDelay(true);
@@ -97,6 +103,7 @@ public DashboardServer(InetSocketAddress address, Logger logger, Kernel root, De
         this.logger.atInfo().log("Starting dashboard server on address: {}", address);
         this.pubSubIPCAgent = pubSubIPCAgent;
         this.mqttClient = mqttClient;
+        this.streamManagerHelper = streamManagerHelper;
     }
 
     // links the API impl and starts the socket server
@@ -251,6 +258,40 @@ public void onMessage(WebSocket conn, String msg) {
                     unsubscribeFromPubSubTopic(conn, packedRequest, req);
                     break;
                 }
+                case streamManagerListStreams: {
+                    streamManagerListStreams(conn, packedRequest);
+                    break;
+                }
+                case streamManagerDescribeStream: {
+                    streamManagerDescribeStream(conn, packedRequest, req);
+                    break;
+                }
+
+                case streamManagerDeleteMessageStream: {
+                    streamManagerDeleteMessageStream(conn, packedRequest, req);
+                    break;
+                }
+
+                case streamManagerReadMessages: {
+                    streamManagerReadMessages(conn, packedRequest, req);
+                    break;
+                }
+
+                case streamManagerAppendMessage:{
+                    streamManagerAppendMessage(conn, packedRequest, req);
+                    break;
+                }
+
+                case streamManagerCreateMessageStream:{
+                    streamManagerCreateMessageStream(conn, packedRequest, req);
+                    break;
+                }
+
+                case streamManagerUpdateMessageStream:{
+                    streamManagerUpdateMessageStream(conn, packedRequest, req);
+                    break;
+                }
+
                 default: { // echo
                     sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, req.call));
                     break;
@@ -355,6 +396,111 @@ private void publishToPubSubTopic(WebSocket conn, PackedRequest packedRequest, R
         }
     }
 
+    private void streamManagerListStreams(WebSocket conn, PackedRequest packedRequest) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            responseMessage.streamsList = this.streamManagerHelper.listStreams();
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while listing streams:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerDescribeStream(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            responseMessage.messageStreamInfo = this.streamManagerHelper.describeStream(req.args[0]);
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while describing stream:",e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerDeleteMessageStream(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            this.streamManagerHelper.deleteMessageStream(req.args[0]);
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while deleting stream:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerReadMessages(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            if (req.args.length == 5) {
+                responseMessage.messagesList = this.streamManagerHelper.readMessages(
+                                                    req.args[0],
+                                                    Long.parseLong(req.args[1]),
+                                                    Long.parseLong(req.args[2]),
+                                                    Long.parseLong(req.args[3]),
+                                                    Long.parseLong(req.args[4]));
+                responseMessage.successful = true;
+            }
+            else{
+                logger.atError().log("StreamManagerReadMessages requires 5 arguments");
+                responseMessage.errorMsg = "StreamManagerReadMessages requires 5 arguments";
+            }
+        }
+        catch (Exception e){
+            logger.error("Error while reading messages:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerAppendMessage(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            this.streamManagerHelper.appendMessage(req.args[0], req.args[1].getBytes());
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while appending message to the stream:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerCreateMessageStream(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            MessageStreamDefinition messageStreamDefinition = jsonMapper.readValue(req.args[0], MessageStreamDefinition.class);
+            this.streamManagerHelper.createMessageStream(messageStreamDefinition);
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while appending message to the stream:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+    private void streamManagerUpdateMessageStream(WebSocket conn, PackedRequest packedRequest, Request req) {
+        StreamManagerResponseMessage responseMessage = new StreamManagerResponseMessage();
+        try {
+            MessageStreamDefinition messageStreamDefinition = jsonMapper.readValue(req.args[0], MessageStreamDefinition.class);
+            this.streamManagerHelper.updateMessageStream(messageStreamDefinition);
+            responseMessage.successful = true;
+        }
+        catch (Exception e){
+            logger.error("Error while appending message to the stream:", e);
+            responseMessage.errorMsg = Utils.generateFailureMessage(e);
+        }
+        sendIfOpen(conn, new Message(MessageType.RESPONSE, packedRequest.requestID, responseMessage));
+    }
+
+
     @Override
     public void onClose(WebSocket conn, int code, String reason, boolean remote) {
         connections.remove(conn);

src/main/java/com/aws/greengrass/localdebugconsole/SimpleHttpServer.java

@@ -12,6 +12,7 @@
 import com.aws.greengrass.dependency.ImplementsService;
 import com.aws.greengrass.dependency.State;
 import com.aws.greengrass.deployment.DeviceConfiguration;
+import com.aws.greengrass.ipc.AuthenticationHandler;
 import com.aws.greengrass.lifecyclemanager.Kernel;
 import com.aws.greengrass.lifecyclemanager.PluginService;
 import com.aws.greengrass.logging.api.Logger;
@@ -96,6 +97,7 @@
 import javax.net.ssl.SSLException;
 
 import static com.aws.greengrass.componentmanager.KernelConfigResolver.CONFIGURATION_CONFIG_KEY;
+import static com.aws.greengrass.ipc.AuthenticationHandler.SERVICE_UNIQUE_ID_KEY;
 import static com.aws.greengrass.util.Utils.isEmpty;
 import static io.netty.buffer.Unpooled.copiedBuffer;
 
@@ -131,6 +133,7 @@ public class SimpleHttpServer extends PluginService implements Authenticator {
     private boolean httpsEnabled = DEFAULT_HTTPS_ENABLED;
     private SslContext context;
     private Provider<SSLEngine> engineProvider;
+    private String streamManagerAuthToken;
 
     @Inject
     public SimpleHttpServer(Topics t, Kernel kernel, DeviceConfiguration deviceConfiguration) {
@@ -142,6 +145,10 @@ public SimpleHttpServer(Topics t, Kernel kernel, DeviceConfiguration deviceConfi
     @Override
     public void postInject() {
         super.postInject();
+        // Does not happen for built-in/plugin services so doing explicitly
+        AuthenticationHandler.registerAuthenticationToken(this);
+        streamManagerAuthToken = Coerce.toString(this.getPrivateConfig().findLeafChild(SERVICE_UNIQUE_ID_KEY));
+
         config.lookup(CONFIGURATION_CONFIG_KEY, "port").dflt(port).subscribe((w, n) -> {
             int oldPort = port;
             port = Coerce.toInt(n);
@@ -195,7 +202,7 @@ public void startup() throws InterruptedException {
 
         logger.atInfo().log("Starting local dashboard server");
         dashboardServer = new DashboardServer(new InetSocketAddress(bindHostname, websocketPort), logger,
-                kernel, deviceConfig, this, engineProvider);
+                kernel, deviceConfig, this, engineProvider, streamManagerAuthToken);
         dashboardServer.startup();
         try {
             // We need to wait for the server to startup before grabbing the port because it starts in a separate thread
@@ -531,6 +538,10 @@ private boolean authenticated(String authHeader) {
 
     @Override
     public boolean isUsernameAndPasswordValid(Pair<String, String> usernameAndPassword) {
+        if (consoleAuthDisabled()) {
+            return true;
+        }
+
         Topics passwordTopics = config.getRoot().findTopics(DEBUG_PASSWORD_NAMESPACE);
         if (passwordTopics == null || usernameAndPassword == null) {
             return false;
@@ -563,7 +574,15 @@ public boolean isUsernameAndPasswordValid(Pair<String, String> usernameAndPasswo
         return Instant.now().isBefore(Instant.ofEpochMilli(Coerce.toLong(expirationTopic)));
     }
 
+    private static boolean consoleAuthDisabled() {
+        // Set to true in development so you don't need to authenticate all the time.
+        return false;
+    }
+
     private Pair<String, String> getUsernameAndPassword(String authHeader) {
+        if (consoleAuthDisabled()) {
+            return new Pair<>("", "");
+        }
         if (Utils.isEmpty(authHeader)) {
             return null;
         }