Add ironfish rpc:status command (iron-fish#1600)

Co-authored-by: Jason Spafford <[email protected]>

ironfish-cli/src/commands/rpc/status.ts

@@ -0,0 +1,77 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+import { FileUtils, GetRpcStatusResponse, PromiseUtils } from '@ironfish/sdk'
+import { Flags } from '@oclif/core'
+import blessed from 'blessed'
+import { IronfishCommand } from '../../command'
+import { RemoteFlags } from '../../flags'
+
+export default class Status extends IronfishCommand {
+  static description = 'Show the status of the RPC layer'
+
+  static flags = {
+    ...RemoteFlags,
+    follow: Flags.boolean({
+      char: 'f',
+      default: false,
+      description: 'follow the status of the node live',
+    }),
+  }
+
+  async start(): Promise<void> {
+    const { flags } = await this.parse(Status)
+
+    if (!flags.follow) {
+      const client = await this.sdk.connectRpc()
+      const response = await client.getRpcStatus()
+      this.log(renderStatus(response.content))
+      this.exit(0)
+    }
+
+    // Console log will create display issues with Blessed
+    this.logger.pauseLogs()
+
+    const screen = blessed.screen({ smartCSR: true })
+    const statusText = blessed.text()
+    screen.append(statusText)
+
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      const connected = await this.sdk.client.tryConnect()
+
+      if (!connected) {
+        statusText.clearBaseLine(0)
+        statusText.setContent('Node: STOPPED')
+        screen.render()
+        await PromiseUtils.sleep(1000)
+        continue
+      }
+
+      const response = this.sdk.client.getRpcStatusStream()
+      for await (const value of response.contentStream()) {
+        statusText.setContent(renderStatus(value))
+        screen.render()
+      }
+    }
+  }
+}
+
+function renderStatus(content: GetRpcStatusResponse): string {
+  let result = `STARTED: ${String(content.started)}`
+
+  for (const adapter of content.adapters) {
+    result += `\n\n[${adapter.name}]\n`
+    result += `Clients:          ${adapter.clients}\n`
+    result += `Requests Pending: ${adapter.pending}\n`
+    result += `Inbound Traffic:  ${FileUtils.formatMemorySize(adapter.inbound)}/s\n`
+    result += `Outbound Traffic: ${FileUtils.formatMemorySize(adapter.outbound)}/s\n`
+    result += `Outbound Total:   ${FileUtils.formatMemorySize(adapter.writtenBytes)}\n`
+    result += `Inbound Total:    ${FileUtils.formatMemorySize(adapter.readBytes)}\n`
+    result += `RW Backlog:       ${FileUtils.formatMemorySize(
+      adapter.readableBytes,
+    )} / ${FileUtils.formatMemorySize(adapter.writableBytes)}`
+  }
+
+  return result
+}

ironfish/src/rpc/adapters/ipcAdapter.ts

@@ -7,6 +7,7 @@ import { v4 as uuid } from 'uuid'
 import * as yup from 'yup'
 import { Assert } from '../../assert'
 import { createRootLogger, Logger } from '../../logger'
+import { Meter } from '../../metrics/meter'
 import { YupUtils } from '../../utils/yup'
 import { Request } from '../request'
 import { ApiNamespace, Router } from '../routes'
@@ -88,6 +89,8 @@ export class RpcIpcAdapter implements IRpcAdapter {
   pending = new Map<IpcSocketId, Request[]>()
   started = false
   connection: IpcAdapterConnectionInfo
+  inboundTraffic = new Meter()
+  outboundTraffic = new Meter()
 
   constructor(
     namespaces: ApiNamespace[],
@@ -105,6 +108,9 @@ export class RpcIpcAdapter implements IRpcAdapter {
     }
     this.started = true
 
+    this.inboundTraffic.start()
+    this.outboundTraffic.start()
+
     const { IPC } = await import('node-ipc')
     const ipc = new IPC()
     ipc.config.silent = true
@@ -152,6 +158,9 @@ export class RpcIpcAdapter implements IRpcAdapter {
   }
 
   async stop(): Promise<void> {
+    this.inboundTraffic.stop()
+    this.outboundTraffic.stop()
+
     if (this.started && this.ipc) {
       this.ipc.server.stop()
 
@@ -224,6 +233,8 @@ export class RpcIpcAdapter implements IRpcAdapter {
       return
     }
 
+    this.inboundTraffic.add(Buffer.from(JSON.stringify(data)).byteLength)
+
     const message = result.result
     const router = this.router
     const server = this.server
@@ -263,11 +274,13 @@ export class RpcIpcAdapter implements IRpcAdapter {
   emitResponse(socket: IpcSocket, messageId: number, status: number, data: unknown): void {
     Assert.isNotNull(this.server)
     this.server.emit(socket, 'message', { id: messageId, status: status, data: data })
+    this.outboundTraffic.add(Buffer.from(JSON.stringify(data)).byteLength)
   }
 
   emitStream(socket: IpcSocket, messageId: number, data: unknown): void {
     Assert.isNotNull(this.server)
     this.server.emit(socket, 'stream', { id: messageId, data: data })
+    this.outboundTraffic.add(Buffer.from(JSON.stringify(data)).byteLength)
   }
 
   renderError(error: Error): IpcError {
@@ -296,5 +309,6 @@ export class RpcIpcAdapter implements IRpcAdapter {
     }
 
     this.server.emit(socket, 'malformedRequest', error)
+    this.outboundTraffic.add(Buffer.from(JSON.stringify(error)).byteLength)
   }
 }

ironfish/src/rpc/adapters/socketAdapter/socketAdapter.ts

@@ -4,6 +4,7 @@
 import net from 'net'
 import { v4 as uuid } from 'uuid'
 import { createRootLogger, Logger } from '../../../logger'
+import { Meter } from '../../../metrics/meter'
 import { JSONUtils } from '../../../utils'
 import { ErrorUtils } from '../../../utils/error'
 import { YupUtils } from '../../../utils/yup'
@@ -38,6 +39,9 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
   started = false
   clients = new Map<string, SocketClient>()
 
+  inboundTraffic = new Meter()
+  outboundTraffic = new Meter()
+
   constructor(
     host: string,
     port: number,
@@ -61,6 +65,9 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
     const server = await this.createServer()
     this.server = server
 
+    this.inboundTraffic.start()
+    this.outboundTraffic.start()
+
     return new Promise((resolve, reject) => {
       server.on('error', (err) => {
         reject(err)
@@ -84,6 +91,9 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
       return
     }
 
+    this.inboundTraffic.stop()
+    this.outboundTraffic.stop()
+
     this.clients.forEach((client) => {
       client.requests.forEach((r) => r.close())
       client.socket.destroy()
@@ -153,7 +163,9 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
   }
 
   async onClientData(client: SocketClient, data: Buffer): Promise<void> {
+    this.inboundTraffic.add(data.byteLength)
     client.messageBuffer.write(data)
+
     for (const rpcMessage of client.messageBuffer.readMessages()) {
       const [parsed, error] = JSONUtils.tryParse(rpcMessage)
       if (error) {
@@ -209,6 +221,7 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
   emitResponse(client: SocketClient, data: ServerSocketRpc, requestId?: string): void {
     const message = this.encodeNodeIpc(data)
     client.socket.write(message)
+    this.outboundTraffic.add(message.byteLength)
 
     if (requestId) {
       client.requests.get(requestId)?.close()
@@ -219,15 +232,16 @@ export abstract class RpcSocketAdapter implements IRpcAdapter {
   emitStream(client: SocketClient, data: ServerSocketRpc): void {
     const message = this.encodeNodeIpc(data)
     client.socket.write(message)
+    this.outboundTraffic.add(message.byteLength)
   }
 
   // `constructResponse`,  `constructStream` and `constructMalformedRequest` construct messages to return
   // to a 'node-ipc' client. Once we remove 'node-ipc' we can return our own messages
   // The '\f' is for handling the delimeter that 'node-ipc' expects when parsing
   // messages it received. See 'node-ipc' parsing/formatting logic here:
   // https://github.com/RIAEvangelist/node-ipc/blob/master/entities/EventParser.js
-  encodeNodeIpc(ipcResponse: ServerSocketRpc): string {
-    return JSON.stringify(ipcResponse) + MESSAGE_DELIMITER
+  encodeNodeIpc(ipcResponse: ServerSocketRpc): Buffer {
+    return Buffer.from(JSON.stringify(ipcResponse) + MESSAGE_DELIMITER)
   }
 
   constructMessage(messageId: number, status: number, data: unknown): ServerSocketRpc {

ironfish/src/rpc/clients/client.ts

@@ -72,6 +72,7 @@ import {
   GetPeerMessagesRequest,
   GetPeerMessagesResponse,
 } from '../routes/peers/getPeerMessages'
+import { GetRpcStatusRequest, GetRpcStatusResponse } from '../routes/rpc/getStatus'
 
 export abstract class RpcClient {
   readonly logger: Logger
@@ -257,6 +258,24 @@ export abstract class RpcClient {
     })
   }
 
+  async getRpcStatus(
+    params: GetRpcStatusRequest = undefined,
+  ): Promise<ResponseEnded<GetRpcStatusResponse>> {
+    return this.request<GetRpcStatusResponse>(
+      `${ApiNamespace.rpc}/getStatus`,
+      params,
+    ).waitForEnd()
+  }
+
+  getRpcStatusStream(
+    params: GetRpcStatusRequest = undefined,
+  ): Response<void, GetRpcStatusResponse> {
+    return this.request<void, GetRpcStatusResponse>(`${ApiNamespace.rpc}/getStatus`, {
+      ...params,
+      stream: true,
+    })
+  }
+
   onGossipStream(params: OnGossipRequest = undefined): Response<void, OnGossipResponse> {
     return this.request<void, OnGossipResponse>(`${ApiNamespace.event}/onGossip`, params)
   }

ironfish/src/rpc/routes/index.ts

@@ -8,6 +8,7 @@ export * from './events'
 export * from './node'
 export * from './peers'
 export * from './router'
+export * from './rpc'
 export * from './mining'
 export * from './transactions'
 export * from './faucet'

ironfish/src/rpc/routes/router.ts

@@ -22,6 +22,7 @@ export enum ApiNamespace {
   transaction = 'transaction',
   telemetry = 'telemetry',
   worker = 'worker',
+  rpc = 'rpc',
 }
 
 export const ALL_API_NAMESPACES = StrEnumUtils.getValues(ApiNamespace)