Fix issues with ElastiCache deployments (#249)

* Make server host and port configurable through env variables

* Fix lint issues

Signed-off-by: Ravjot Brar <Ravjot.Brar@VAN-FM2FQ6NHXT.local>

* Remove window from metrics server

Signed-off-by: Ravjot Brar <ravjot.brar@improving.com>

* Replace host and port vars with ws_url

* Get frontend to infer ws protocol

* Changed name of ws env variable

* Add alternatives to CONFIG commands

* Make sure url doesn't rely on checking if browser

* Change name of ws_url variable

* Add cluster slot stats warning

* Fix ws url generation

* Comment out enableClusterSlotStats

* MODULE also restricted

* Add fix for idle timeout

* Add interval time

* Address review and remove unnecessary comments

---------

Signed-off-by: Ravjot Brar <Ravjot.Brar@VAN-FM2FQ6NHXT.local>
Signed-off-by: Ravjot Brar <ravjot.brar@improving.com>
Co-authored-by: Ravjot Brar <Ravjot.Brar@VAN-FM2FQ6NHXT.local>

Author: ravjotbrar

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

@@ -343,7 +343,7 @@ export const getHotKeysEpic = (store: Store) =>
       const state = store.getState()
       const connection = state.valkeyConnection.connections[connectionId]
       const monitorEnabled = state.config[connectionId].monitoring.monitorEnabled
-      const lfuEnabled = connection.connectionDetails.keyEvictionPolicy.includes("lfu") ?? false
+      const lfuEnabled = connection.connectionDetails.keyEvictionPolicy?.includes("lfu") ?? false
       const clusterSlotStatsEnabled = connection.connectionDetails.clusterSlotStatsEnabled ?? false
 
       socket.next({

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

@@ -26,13 +26,14 @@ import {
 
 let socket$: WebSocketSubject<PayloadAction> | null = null
 
-const isBrowser = typeof window !== "undefined"
+const isElectron = window.location.protocol === "file:"
+const isHttps = window.location.protocol === "https:"
 
 const url =
-  process.env.WS_URL ||
-  (isBrowser
-    ? `ws://${window.location.hostname}:${window.location.port || "8080"}`
-    : "ws://localhost:8080")
+  process.env.VALKEY_ADMIN_WS_URL ||
+  (isElectron
+    ? "ws://localhost:8080"
+    : `${isHttps ? "wss" : "ws"}://${window.location.host}`)
 
 const connect = (store: Store) =>
   action$.pipe(

apps/server/src/__tests__/connection.test.ts

@@ -144,7 +144,10 @@ describe("connectToValkey", () => {
         ) {
           return [{ key: "maxmemory-policy", value: "allkeys-lfu" }]
         }
-
+        else if (  
+          Array.isArray(args) &&
+          args[0] === "JSON.TYPE"
+        ) throw Error
         // default response for other commands
         return []
       }),
@@ -301,9 +304,7 @@ describe("connectToValkey", () => {
 
   it("should return false when JSON module is not present", async () => {
     const mockClient = {
-      customCommand: mock.fn(async () => [
-        [{ key: "name", value: "search" }],
-      ]),
+      customCommand: mock.fn(async () => { throw Error }),
     }
 
     const result = await checkJsonModuleAvailability(mockClient as any)

apps/server/src/check-json-module.ts

@@ -1,27 +1,14 @@
 import { GlideClient, GlideClusterClient } from "@valkey/valkey-glide"
 
-type ModuleAttribute = {
-  key: string
-  value: unknown
-}
-
-type Module = ModuleAttribute[]
-
 export async function checkJsonModuleAvailability(
   client: GlideClient | GlideClusterClient,
 ): Promise<boolean> {
   try {
-    const modules = await client.customCommand(["MODULE", "LIST"]) as Module[]
-
-    return modules.some((module) => {
-      const moduleName = module.find((attr) => attr.key === "name")?.value
-
-      return typeof moduleName === "string" &&
-             (moduleName.toLowerCase().includes("json") ||
-              moduleName.toLowerCase().includes("rejson"))
-    })
-  } catch (err) {
-    console.error("Error checking JSON module availability:", err)
+    // Elasticache restricts MODULE command
+    await client.customCommand(["JSON.TYPE", "nonexistent_key"])
+    return true
+  } catch {
     return false
   }
 }
+

apps/server/src/connection.ts

@@ -61,8 +61,24 @@ export async function connectToValkey(
     // Only start metrics server if it hasn't been started before
     if (!metricsServerMap.has(payload.connectionId)) await startMetricsServer(payload.connectionDetails, payload.connectionId)
 
-    const evictionPolicyResponse = await standaloneClient.customCommand(["CONFIG", "GET", "maxmemory-policy"]) as [{key: string, value: string}]
-    const keyEvictionPolicy: KeyEvictionPolicy = evictionPolicyResponse[0].value.toLowerCase() as KeyEvictionPolicy
+    let keyEvictionPolicy: KeyEvictionPolicy = "noeviction"
+    try {
+      const evictionPolicyResponse = await standaloneClient.customCommand(
+        ["CONFIG", "GET", "maxmemory-policy"],
+      ) as [{key: string, value: string}]
+
+      keyEvictionPolicy = R.pipe(
+        R.pathOr("noeviction", [0, "value"]),
+        R.toLower,
+      )(evictionPolicyResponse) as KeyEvictionPolicy
+    } catch {
+      console.warn("Command \"CONFIG\" not available. Trying \"INFO SERVER\" instead")
+      const infoResponse = await standaloneClient.info([InfoOptions.Server])
+      const parsed = parseInfo(infoResponse)
+      if (parsed["maxmemory_policy"]) {
+        keyEvictionPolicy = parsed["maxmemory_policy"].toLowerCase() as KeyEvictionPolicy
+      }
+    }
     const jsonModuleAvailable = await checkJsonModuleAvailability(standaloneClient)
 
     if (await belongsToCluster(standaloneClient)) {
@@ -185,7 +201,6 @@ async function connectToCluster(
   jsonModuleAvailable: boolean,
   clusterNodesMap: Map<string, string[]>,
 ) {
-  await standaloneClient.customCommand(["CONFIG", "SET", "cluster-announce-hostname", addresses[0].host])
   const { clusterNodes, clusterId } = await discoverCluster(standaloneClient, payload)
   if (R.isEmpty(clusterNodes)) {
     throw new Error("No cluster nodes discovered")
@@ -241,10 +256,13 @@ async function connectToCluster(
     clusterNodesMap.set(clusterId, [payload.connectionId])
   }
 
-  const clusterSlotStatsResponse = await clusterClient.customCommand(
-    ["CONFIG", "GET", "cluster-slot-stats-enabled"],
-  ) as [Record<string, string>]
-  const clusterSlotStatsEnabled = clusterSlotStatsResponse[0].value === "yes" 
+  let clusterSlotStatsEnabled = false
+  try {
+    await clusterClient.customCommand(["CLUSTER", "SLOT-STATS", "SLOTSRANGE", "0", "0"])
+    clusterSlotStatsEnabled = true
+  } catch {
+    console.warn("Cluster slot-stats is not enabled.")
+  }
 
   const clusterConnectionInfo = {
     type: VALKEY.CONNECTION.clusterConnectFulfilled,

apps/server/src/index.ts

@@ -34,6 +34,10 @@ import {
 } from "./metrics-orchestrator"
 import type { Request, Response } from "express"
 
+interface AliveWebSocket extends WebSocket {
+  isAlive: boolean
+}
+
 interface MetricsServerMessage {
   type: string
   payload: {
@@ -102,8 +106,18 @@ server.listen(port, () => {
   }
 })
 
-wss.on("connection", (ws: WebSocket) => {
+const interval = setInterval(() => {
+  wss.clients.forEach((ws) => {
+    const aliveSocket = ws as AliveWebSocket
+    if (aliveSocket.isAlive === false) return ws.terminate()
+    aliveSocket.isAlive = false
+    ws.ping()
+  })
+}, 30000)
+wss.on("connection", (ws: AliveWebSocket) => {
   console.log("Client connected.")
+  ws.isAlive = true
+  ws.on("pong", () => {ws.isAlive = true})
   // This is a simplified cluster node map that stores clusterIds and their corresponding nodeIds
   const clusterNodesMap: Map<string, string[]> = new Map()
 
@@ -191,7 +205,7 @@ wss.on("connection", (ws: WebSocket) => {
 
 function shutdown() {
   console.log("Shutdown signal received")
-
+  clearInterval(interval)
   // Close websocket clients
   wss.clients.forEach((ws) => {
     try {