feat(cli): add logs command

.changeset/three-rings-serve.md

@@ -0,0 +1,7 @@
+---
+"@plutolang/pulumi-adapter": patch
+"@plutolang/pluto-infra": patch
+"@plutolang/cli": patch
+---
+
+feat(cli): add `logs` command

apps/cli/package.json

@@ -31,6 +31,7 @@
     "lint": "eslint ."
   },
   "dependencies": {
+    "@aws-sdk/client-cloudwatch-logs": "^3.583.0",
     "@aws-sdk/client-lambda": "^3.462.0",
     "@inquirer/prompts": "^3.2.0",
     "@plutolang/base": "workspace:^",

apps/cli/src/commands/index.ts

@@ -5,3 +5,4 @@ export { deploy } from "./deploy";
 export { destroy } from "./destory";
 export { test } from "./test";
 export { newStack } from "./stack_new";
+export { log } from "./log";

apps/cli/src/commands/log.ts

@@ -0,0 +1,162 @@
+import chalk from "chalk";
+import {
+  CloudWatchLogsClient,
+  FilterLogEventsCommand,
+  FilterLogEventsCommandInput,
+  ResourceNotFoundException,
+} from "@aws-sdk/client-cloudwatch-logs";
+import { PlatformType, ProvisionType, core } from "@plutolang/base";
+import logger from "../log";
+import { getStackBasicDirs } from "../utils";
+import {
+  buildAdapterByProvisionType,
+  loadArchRef,
+  loadProjectAndStack,
+  loadProjectRoot,
+} from "./utils";
+
+const CHALK_COLOR_FNS = [
+  chalk.red,
+  chalk.green,
+  chalk.yellow,
+  chalk.blue,
+  chalk.magenta,
+  chalk.cyan,
+  chalk.white,
+  chalk.gray,
+  chalk.redBright,
+  chalk.greenBright,
+  chalk.yellowBright,
+  chalk.blueBright,
+  chalk.magentaBright,
+  chalk.cyanBright,
+  chalk.whiteBright,
+];
+
+export interface LogOptions {
+  stack?: string;
+  follow?: boolean;
+  all?: boolean;
+  showPlatform?: boolean;
+}
+
+export async function log(opts: LogOptions) {
+  try {
+    const projectRoot = loadProjectRoot();
+    const { project, stack } = loadProjectAndStack(projectRoot, opts.stack);
+
+    if (stack.provisionType !== ProvisionType.Pulumi || stack.platformType !== PlatformType.AWS) {
+      logger.error("Currently, only pulumi stacks on AWS are supported.");
+      return;
+    }
+
+    if (!stack.archRefFile || !stack.provisionFile) {
+      throw new Error(
+        "The stack is missing an architecture reference file and a provision file. Please execute the `pluto deploy` command again before proceeding with the log command."
+      );
+    }
+
+    // Get the deployed resources from Pulumi adapter
+    const { stateDir } = getStackBasicDirs(projectRoot, stack.name);
+    const adapter = await buildAdapterByProvisionType(stack.provisionType, {
+      project: project.name,
+      rootpath: project.rootpath,
+      language: project.language,
+      stack: stack,
+      archRef: loadArchRef(stack.archRefFile),
+      entrypoint: stack.provisionFile,
+      stateDir: stateDir,
+    });
+    const result = await adapter.state();
+    await viewLogForAWS(result.instances, opts);
+  } catch (e) {
+    if (e instanceof Error) {
+      logger.error("Failed to view the log: ", e.message);
+    } else {
+      logger.error("Failed to view the log: ", e);
+    }
+    logger.debug(e);
+    process.exit(1);
+  }
+}
+
+/**
+ * View the logs for the AWS stack
+ * @param resources The deployed resources in the stack
+ * @param options The options for the log command
+ */
+async function viewLogForAWS(resources: core.ResourceInstance[], options?: LogOptions) {
+  // Get the names of the Lambda functions, and then print their logs
+  const AWS_LAMBDA_TYPE_IN_PULUMI = "aws:lambda/function:Function";
+  const promises = resources
+    .filter((ins) => ins.type === AWS_LAMBDA_TYPE_IN_PULUMI)
+    .map((lambda, idx) => {
+      const name = lambda.name;
+      return viewLogOfLambda(name, idx);
+    });
+  if (promises.length === 0) {
+    logger.warn("No functions found in the stack");
+  }
+  await Promise.all(promises);
+
+  async function viewLogOfLambda(lambdaName: string, idx: number) {
+    logger.info("View the logs for Lambda function: ", lambdaName);
+
+    const client = new CloudWatchLogsClient();
+    const logGroupName = `/aws/lambda/${lambdaName}`;
+    const colorFn = CHALK_COLOR_FNS[idx % CHALK_COLOR_FNS.length];
+
+    // Get logs within the last 5 minutes if  not all logs are requested
+    const startTime = options?.all ? 0 : new Date().getTime() - 5 * 60 * 1000;
+    let nextToken: string | undefined;
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      try {
+        const params: FilterLogEventsCommandInput = {
+          logGroupName,
+          startTime,
+          nextToken,
+        };
+        const command = new FilterLogEventsCommand(params);
+        const response = await client.send(command);
+
+        response.events?.forEach((event) => {
+          if (!options?.showPlatform && isPlatformLog(event.message ?? "")) {
+            // If the log message is a platform log, and the user does not want to show them, skip
+            // it
+            return;
+          }
+
+          const formattedTime = new Date(event.timestamp!).toISOString();
+          console.log(`${colorFn(lambdaName)} | ${formattedTime} | ${event.message?.trim()}`);
+        });
+
+        // Update the token for the next iteration
+        nextToken = response.nextToken ?? nextToken;
+      } catch (e) {
+        if (e instanceof ResourceNotFoundException) {
+          logger.debug(`No logs found for Lambda function: ${lambdaName}`);
+        } else {
+          throw e;
+        }
+      }
+
+      if (!options?.follow) {
+        // If not following, exit the loop
+        break;
+      } else {
+        // Wait for 500 milliseconds before fetching the next batch of logs
+        await new Promise((resolve) => setTimeout(resolve, 500));
+      }
+    }
+  }
+
+  function isPlatformLog(logMessage: string) {
+    if (
+      /^INIT_START|^START RequestId|^END RequestId|^INIT_REPORT|^REPORT RequestId/g.test(logMessage)
+    ) {
+      return true;
+    }
+    return false;
+  }
+}

apps/cli/src/log.ts

@@ -2,16 +2,20 @@ import chalk from "chalk";
 
 class Logger {
   public static error(...msg: any[]) {
-    console.error(chalk.bold.red("Error:"), ...msg);
+    console.error(chalk.bold.red("ERRO:"), ...msg);
+  }
+
+  public static warn(...msg: any[]) {
+    console.warn(chalk.yellow("WARN:"), ...msg);
   }
 
   public static info(...msg: any[]) {
-    console.info(chalk.blue("Info: "), ...msg);
+    console.info(chalk.blue("INFO: "), ...msg);
   }
 
   public static debug(...msg: any[]) {
     if (process.env.DEBUG) {
-      console.debug(chalk.gray("Debug:"), ...msg);
+      console.debug(chalk.gray("DEBU:"), ...msg);
     }
   }
 }

apps/cli/src/main.ts

@@ -115,6 +115,17 @@ async function main() {
     )
     .action(cmd.destroy);
 
+  program
+    .command("logs")
+    .description(
+      "View the logs of the application. Please be aware that the logs that are displayed may not be in sequence. "
+    )
+    .option("-s, --stack <stack>", "Specified stack")
+    .option("-f, --follow", "Follow the log, like `tail -f`", false)
+    .option("-a, --all", "Show all logs from the beginning", false)
+    .option("--show-platform", "Show the logs of the platform", false)
+    .action(cmd.log);
+
   program.command("stack", "Manage stacks");
 
   if (process.env["DEBUG"]) {

components/adapters/pulumi/src/pulumi.ts

@@ -89,7 +89,7 @@ export class PulumiAdapter extends core.Adapter {
         throw new Error("Cannot find the target pulumi stack. Have you already deployed it?");
       }
 
-      await pulumiStack.refresh({ onOutput: debugStream });
+      // await pulumiStack.refresh({ onOutput: debugStream });
       const result = await pulumiStack.exportStack();
 
       const instances: core.ResourceInstance[] = [];

packages/pluto-infra/src/aws/apigateway_adapter.py

@@ -21,7 +21,8 @@ def handler(event, context):
         query=event.get("queryStringParameters", {}) or {},
         body=payload,
     )
-    print("Pluto: Handling HTTP request: ", request)
+    if os.environ.get("DEBUG"):
+        print("Pluto: Handling HTTP request: ", request)
 
     try:
         user_handler: Callable[[HttpRequest], HttpResponse] = globals()["__handler_"]

packages/pluto-infra/src/aws/common_top_adapter.py

@@ -7,18 +7,21 @@
 depth = 0
 child_directory = os.path.join(current_directory, f"child_{depth}")
 while os.path.exists(child_directory):
-    print(f"Adding {child_directory} to the system path.")
+    if os.environ.get("DEBUG"):
+        print(f"Adding {child_directory} to the system path.")
     sys.path.insert(0, child_directory)
 
     site_pkgs_dir = os.path.join(child_directory, "site-packages")
     if os.path.exists(site_pkgs_dir):
-        print(f"Adding {site_pkgs_dir} to the system path.")
+        if os.environ.get("DEBUG"):
+            print(f"Adding {site_pkgs_dir} to the system path.")
         sys.path.insert(0, site_pkgs_dir)
 
     depth += 1
     child_directory = os.path.join(child_directory, f"child_{depth}")
 
-print("The system path is:", sys.path)
+if os.environ.get("DEBUG"):
+    print("The system path is:", sys.path)
 
 
 def handler(*args, **kwargs):

packages/pluto-infra/src/aws/lambda_adapter.py

@@ -18,7 +18,8 @@ def handler(payload, context):
     account_id = context.invoked_function_arn.split(":")[4]
     os.environ["AWS_ACCOUNT_ID"] = account_id
     try:
-        print("Payload:", payload)
+        if os.environ.get("DEBUG"):
+            print("Payload:", payload)
         if is_http_payload(payload):
             if payload["isBase64Encoded"]:
                 body = base64.b64decode(payload["body"]).decode("utf-8")

packages/pluto-infra/src/aws/sns_subscriber_adapter.py

@@ -15,7 +15,8 @@ def handler(event, context):
         payload = record["Sns"]["Message"]
         data = json.loads(payload)
         cloud_event = CloudEvent(timestamp=data["timestamp"], data=data["data"])
-        print("Pluto: Handling event: ", cloud_event)
+        if os.environ.get("DEBUG"):
+            print("Pluto: Handling event: ", cloud_event)
 
         try:
             user_handler: Callable[[CloudEvent], None] = globals()["__handler_"]