Integrate hotkeys endpoint with backend

apps/frontend/src/state/epics/rootEpic.ts

@@ -1,7 +1,15 @@
 import { merge } from "rxjs"
 import { wsConnectionEpic } from "./wsEpics"
-import { connectionEpic, sendRequestEpic, setDataEpic, deleteConnectionEpic, autoReconnectEpic, valkeyRetryEpic, 
-  updateConnectionDetailsEpic } from "./valkeyEpics"
+import { 
+  connectionEpic, 
+  sendRequestEpic, 
+  setDataEpic, 
+  deleteConnectionEpic, 
+  updateConnectionDetailsEpic,
+  autoReconnectEpic, 
+  valkeyRetryEpic, 
+  getHotKeysEpic 
+} from "./valkeyEpics"
 import { keyBrowserEpic } from "./keyBrowserEpic"
 import type { Store } from "@reduxjs/toolkit"
 
@@ -15,6 +23,7 @@ export const registerEpics = (store: Store) => {
     updateConnectionDetailsEpic(store),
     sendRequestEpic(),
     setDataEpic(),
+    getHotKeysEpic(),
     keyBrowserEpic(),
   ).subscribe({
     error: (err) => console.error("Epic error:", err),

apps/frontend/src/state/epics/valkeyEpics.ts

@@ -19,6 +19,7 @@ import { setData } from "../valkey-features/info/infoSlice"
 import { action$, select } from "../middleware/rxjsMiddleware/rxjsMiddlware"
 import { setClusterData } from "../valkey-features/cluster/clusterSlice"
 import { connectFulfilled as wsConnectFulfilled } from "../wsconnection/wsConnectionSlice"
+import { hotKeysRequested } from "../valkey-features/hotkeys/hotKeysSlice.ts"
 import history from "../../history.ts"
 import type { Store } from "@reduxjs/toolkit"
 
@@ -249,3 +250,12 @@ export const setDataEpic = () =>
       history.navigate(dashboardPath)
     }),
   )
+
+export const getHotKeysEpic = () => 
+  action$.pipe(
+    select(hotKeysRequested),
+    tap((action) => {
+      const socket = getSocket()
+      socket.next(action)
+    }),
+  )

apps/frontend/src/state/valkey-features/hotkeys/hotKeysSlice.ts

@@ -0,0 +1,60 @@
+import { createSlice } from "@reduxjs/toolkit"
+import { type JSONObject } from "@common/src/json-utils"
+import { VALKEY } from "@common/src/constants.ts"
+import * as R from "ramda"
+import type { RootState } from "@/store.ts"
+
+export const selectHotKeys = (id: string) => (state: RootState) =>
+  R.path([VALKEY.HOTKEYS.name, id, "hotKeys"], state)
+
+interface HotKeysState {
+  [connectionId: string]: {
+    hotKeys: [string, number][]
+    checkAt: string|null,
+    monitorRunning: boolean,
+    nodeId: string|null,
+    error?: JSONObject | null,
+  }
+}
+
+const initialHotKeysState: HotKeysState = {}
+
+const hotKeysSlice = createSlice({
+  name: "hotKeys",
+  initialState: initialHotKeysState,
+  reducers: {
+    hotKeysRequested: (state, action) => {
+      const connectionId = action.payload.connectionId
+      if (!state[connectionId]) {
+        state[connectionId] = {
+          hotKeys: [],
+          checkAt: null, 
+          monitorRunning: false,
+          nodeId: null,
+        }
+      }
+    },
+    hotKeysFulfilled: (state, action) => {
+      const { hotKeys, monitorRunning, checkAt, nodeId } = action.payload.parsedResponse
+      console.log("Parsed response in the frontend is: ", action.payload.parsedResponse)
+      const connectionId = action.payload.connectionId
+      state[connectionId] = {
+        hotKeys,
+        checkAt,
+        monitorRunning, 
+        nodeId,
+      }
+
+    },
+    hotKeysError: (state, action) => {
+      const { connectionId, error } = action.payload
+      state[connectionId].error = error
+    },
+  },
+})
+export default hotKeysSlice.reducer
+export const {
+  hotKeysRequested,
+  hotKeysFulfilled,
+  hotKeysError,
+} = hotKeysSlice.actions

apps/frontend/src/store.ts

@@ -8,6 +8,7 @@ import clusterReducer from "@/state/valkey-features/cluster/clusterSlice"
 import valkeyCommandReducer from "@/state/valkey-features/command/commandSlice.ts"
 import valkeyInfoReducer from "@/state/valkey-features/info/infoSlice.ts"
 import keyBrowserReducer from "@/state/valkey-features/keys/keyBrowserSlice.ts"
+import hotKeysReducer from "@/state/valkey-features/hotkeys/hotKeysSlice.ts"
 
 export const store = configureStore({
   reducer: {
@@ -17,6 +18,7 @@ export const store = configureStore({
     [VALKEY.STATS.name]: valkeyInfoReducer,
     [VALKEY.KEYS.name]: keyBrowserReducer,
     [VALKEY.CLUSTER.name]: clusterReducer,
+    [VALKEY.HOTKEYS.name]: hotKeysReducer,
   },
   middleware: (getDefaultMiddleware) => {
     return getDefaultMiddleware({

apps/metrics/src/analyzers/calculateHotKeys.js

@@ -5,7 +5,6 @@ export const calculateHotKeys = async () => {
   const rows = await Streamer.monitor()
   const ACCESS_COMMANDS = ["get", "set", "mget", "hget", "hgetall", "hmget", "json.get", "json.mget"]
   const CUT_OFF_FREQUENCY = 1
-
   return R.pipe(
     R.reduce((acc, { ts, command }) => {
       const [cmd, ...args] = command.split(' ').filter(Boolean)

apps/metrics/src/effects/ndjson-streamer.js

@@ -1,13 +1,14 @@
 import fs from "node:fs";
 import readline from "node:readline";
 import path from "node:path";
-import { loadConfig } from "../config.js";
+
+const DATA_DIR = process.env.DATA_DIR || path.resolve(process.cwd(), "data")
 
 const dayStr = (date) => date.toISOString().slice(0, 10).replace(/-/g, "");
 
+
 const fileFor = (prefix, date) => {
-  const cfg = loadConfig();
-  const dataDir = cfg.server.data_dir;
+  const dataDir = DATA_DIR
   return path.join(dataDir, `${prefix}_${dayStr(date)}.ndjson`);
 };
 
@@ -17,11 +18,11 @@ export async function streamNdjson(prefix, filterFn = () => true) {
   yesterday.setDate(today.getDate() - 1);
 
   const files = [fileFor(prefix, yesterday), fileFor(prefix, today)];
+
   const results = [];
 
   for (const file of files) {
     if (!fs.existsSync(file)) continue;
-
     const fileStream = fs.createReadStream(file, { encoding: "utf8" });
     const rl = readline.createInterface({ input: fileStream, crlfDelay: Infinity });
 

apps/metrics/src/index.js

@@ -9,7 +9,6 @@ import { MODE, ACTION, MONITOR } from "./utils/constants.js"
 
 async function main() {
   const cfg = loadConfig()
-
   const ensureDir = dir => { if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }) }
   ensureDir(cfg.server.data_dir)
 
@@ -26,7 +25,7 @@ async function main() {
   const client = createClient({ url })
   client.on("error", err => console.error("valkey error", err))
   await client.connect()
-  const stoppers = await setupCollectors(client)
+  const stoppers = await setupCollectors(client, cfg)
 
   const app = express()
 
@@ -82,7 +81,7 @@ async function main() {
           if (monitorRunning) {
             return { monitorRunning }
           }
-          await startMonitor()
+          await startMonitor(cfg)
           monitorRunning = true
           checkAt = Date.now() + monitorDuration
           return { monitorRunning, checkAt}
@@ -122,12 +121,12 @@ async function main() {
         return res.json(monitorResponse)
       }
       if (Date.now() > checkAt) {
-        const hotkeys = await calculateHotKeys()
+        const hotKeys = await calculateHotKeys()
         if (req.query.mode !== MODE.CONTINUOUS) {
           await monitorHandler(ACTION.STOP) 
         }
         monitorResponse = await monitorHandler(ACTION.STATUS)
-        return res.json({ nodeId: url, hotkeys, ...monitorResponse })
+        return res.json({ nodeId: url, hotKeys, ...monitorResponse })
 
       }
       return res.json({ checkAt })

apps/metrics/src/init-collectors.js

@@ -2,13 +2,11 @@ import { makeFetcher } from "./effects/fetchers.js"
 import { makeMonitorStream } from "./effects/monitor-stream.js"
 import { makeNdjsonWriter } from "./effects/ndjson-writer.js"
 import { startCollector } from "./epics/collector-rx.js"
-import { loadConfig } from "./config.js"
 
-const cfg = loadConfig()
 const MONITOR = "monitor"
 const stoppers = {}
 
-const startMonitor = () => {
+const startMonitor = (cfg) => {
   const nd = makeNdjsonWriter({
     dataDir: cfg.server.data_dir,
     filePrefix: MONITOR
@@ -17,7 +15,7 @@ const startMonitor = () => {
   const sink = {
     appendRows: async rows => {
       await nd.appendRows(rows)
-      console.info(`[${monitorEpic.name}] wrote ${rows.length} logs to ${cfg.server.data_dir}/${monitorEpic.file_prefix || monitorEpic.name}`)
+      console.info(`[${monitorEpic.name}] wrote ${rows.length} logs to ${cfg.server.data_dir}/`)
     },
     close: nd.close
   }
@@ -38,7 +36,7 @@ const startMonitor = () => {
 
 const stopMonitor = async () => await stoppers[MONITOR]()
 
-const setupCollectors = async client => {
+const setupCollectors = async (client, cfg) => {
   const fetcher = makeFetcher(client)
   const stoppers = {}
   await Promise.all(cfg.epics

apps/metrics/src/utils/constants.js

@@ -8,4 +8,4 @@ export const ACTION = {
   START: "start",
   STOP: "stop",
   STATUS: "status",
-}
+}

apps/server/src/actions/hotkeys.ts

@@ -0,0 +1,72 @@
+import { type WebSocket } from "ws"
+import { VALKEY } from "../../../../common/src/constants"
+import { withDeps, Deps } from "./utils"
+
+type HotKeysResponse = {
+  nodeId: string
+  hotkeys: [[]]
+  checkAt: number
+  monitorRunning: boolean
+}
+
+const sendHotKeysFulfilled = (
+  ws: WebSocket,
+  connectionId: string,
+  parsedResponse: HotKeysResponse,
+) => {
+  ws.send(
+    JSON.stringify({
+      type: VALKEY.HOTKEYS.hotKeysFulfilled,
+      payload: {
+        connectionId,
+        parsedResponse,
+      },
+    }),
+  )
+}
+
+const sendHotKeysError = (
+  ws: WebSocket,
+  connectionId: string,
+  error: unknown,
+) => {
+  console.log(error)
+  ws.send(
+    JSON.stringify({
+      type: VALKEY.HOTKEYS.hotKeysError,
+      payload: {
+        connectionId,
+        error: error instanceof Error ? error.message : String(error),
+      },
+    }),
+  )
+}
+
+export const hotKeysRequested = withDeps<Deps, void>(
+  async ({ ws, connectionId, metricsServerURIs }) => {
+    const metricsServerURI = metricsServerURIs.get(connectionId)
+    try {
+      const initialResponse = await fetch(`${metricsServerURI}/hot-keys`)
+      const initialParsedResponse: HotKeysResponse = await initialResponse.json() as HotKeysResponse
+      // Initial request starts monitoring and returns when to fetch results (`checkAt`).
+      if (initialParsedResponse.checkAt) {
+        const delay = Math.max(initialParsedResponse.checkAt - Date.now(), 0)
+        // Schedule the follow-up request for when the monitor cycle finishes
+        setTimeout(async () => {
+          try {
+            const dataResponse = await fetch(`${metricsServerURI}/hot-keys`)
+            const dataParsedResponse = await dataResponse.json() as HotKeysResponse
+            sendHotKeysFulfilled(ws, connectionId, dataParsedResponse)
+          } catch (error) {
+            sendHotKeysError(ws, connectionId, error)
+          }
+        }, delay)
+      }
+      else {
+        sendHotKeysFulfilled(ws, connectionId, initialParsedResponse)
+      }
+    } catch (error) {
+      sendHotKeysError(ws, connectionId, error)
+    }
+  },
+)

apps/server/src/actions/utils.ts

@@ -4,7 +4,8 @@ import type WebSocket from "ws"
 export type Deps = {
   ws: WebSocket
   clients: Map<string, GlideClient | GlideClusterClient>
-  connectionId: string
+  connectionId: string,
+  metricsServerURIs: Map<string, string>,
 }
 
 export type ReduxAction = {