Skip to content

Commit eadf1b1

Browse files
Nek-11HenryHengZJ
Nek-11
and
HenryHengZJ
authored
Fix: Add keep-alive options to Redis clients to prevent idle timeouts and socket closing. (#4377)
* redis keepalive mechanism for all redis * removed offline queue commands * Simplified changes for consistency. Added REDIS_KEEP_ALIVE env variable. * update redis socket alive env variable * lint fix --------- Co-authored-by: Henry <[email protected]> Co-authored-by: Henry Heng <[email protected]>
1 parent 98e75ad commit eadf1b1

File tree

14 files changed

+118
-15
lines changed

14 files changed

+118
-15
lines changed

‎docker/.env.example‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -101,4 +101,5 @@ BLOB_STORAGE_PATH=/root/.flowise/storage
101101
# REDIS_CERT=
102102
# REDIS_KEY=
103103
# REDIS_CA=
104-
# ENABLE_BULLMQ_DASHBOARD=
104+
# REDIS_KEEP_ALIVE=
105+
# ENABLE_BULLMQ_DASHBOARD=

‎docker/docker-compose.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ services:
4949
- REDIS_CERT=${REDIS_CERT}
5050
- REDIS_KEY=${REDIS_KEY}
5151
- REDIS_CA=${REDIS_CA}
52+
- REDIS_KEEP_ALIVE=${REDIS_KEEP_ALIVE}
5253
- ENABLE_BULLMQ_DASHBOARD=${ENABLE_BULLMQ_DASHBOARD}
5354
ports:
5455
- '${PORT}:${PORT}'

‎docker/worker/docker-compose.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ services:
4949
- REDIS_CERT=${REDIS_CERT}
5050
- REDIS_KEY=${REDIS_KEY}
5151
- REDIS_CA=${REDIS_CA}
52+
- REDIS_KEEP_ALIVE=${REDIS_KEEP_ALIVE}
5253
- ENABLE_BULLMQ_DASHBOARD=${ENABLE_BULLMQ_DASHBOARD}
5354
ports:
5455
- '${PORT}:${PORT}'

‎packages/components/nodes/cache/RedisCache/RedisCache.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,10 +126,19 @@ const getRedisClient = async (nodeData: INodeData, options: ICommonObject) => {
126126
host,
127127
username,
128128
password,
129+
keepAlive:
130+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
131+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
132+
: undefined,
129133
...tlsOptions
130134
})
131135
} else {
132-
client = new Redis(redisUrl)
136+
client = new Redis(redisUrl, {
137+
keepAlive:
138+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
139+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
140+
: undefined
141+
})
133142
}
134143

135144
return client

‎packages/components/nodes/cache/RedisCache/RedisEmbeddingsCache.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,10 +83,19 @@ class RedisEmbeddingsCache implements INode {
8383
host,
8484
username,
8585
password,
86+
keepAlive:
87+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
88+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
89+
: undefined,
8690
...tlsOptions
8791
})
8892
} else {
89-
client = new Redis(redisUrl)
93+
client = new Redis(redisUrl, {
94+
keepAlive:
95+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
96+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
97+
: undefined
98+
})
9099
}
91100

92101
ttl ??= '3600'

‎packages/components/nodes/memory/RedisBackedChatMemory/RedisBackedChatMemory.ts‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,21 @@ class BufferMemoryExtended extends FlowiseMemory implements MemoryMethods {
132132
}
133133

134134
private async withRedisClient<T>(fn: (client: Redis) => Promise<T>): Promise<T> {
135-
const client = typeof this.redisOptions === 'string' ? new Redis(this.redisOptions) : new Redis(this.redisOptions)
135+
const client =
136+
typeof this.redisOptions === 'string'
137+
? new Redis(this.redisOptions, {
138+
keepAlive:
139+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
140+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
141+
: undefined
142+
})
143+
: new Redis({
144+
...this.redisOptions,
145+
keepAlive:
146+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
147+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
148+
: undefined
149+
})
136150
try {
137151
return await fn(client)
138152
} finally {

‎packages/components/nodes/vectorstores/Redis/Redis.ts‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,15 @@ class Redis_VectorStores implements INode {
147147
}
148148

149149
try {
150-
const redisClient = createClient({ url: redisUrl })
150+
const redisClient = createClient({
151+
url: redisUrl,
152+
socket: {
153+
keepAlive:
154+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
155+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
156+
: undefined // milliseconds
157+
}
158+
})
151159
await redisClient.connect()
152160

153161
const storeConfig: RedisVectorStoreConfig = {
@@ -212,7 +220,15 @@ class Redis_VectorStores implements INode {
212220
redisUrl = 'redis://' + username + ':' + password + '@' + host + ':' + portStr
213221
}
214222

215-
const redisClient = createClient({ url: redisUrl })
223+
const redisClient = createClient({
224+
url: redisUrl,
225+
socket: {
226+
keepAlive:
227+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
228+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
229+
: undefined // milliseconds
230+
}
231+
})
216232

217233
const storeConfig: RedisVectorStoreConfig = {
218234
redisClient: redisClient,

‎packages/server/.env.example‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,4 +99,5 @@ PORT=3000
9999
# REDIS_CERT=
100100
# REDIS_KEY=
101101
# REDIS_CA=
102-
# ENABLE_BULLMQ_DASHBOARD=
102+
# REDIS_KEEP_ALIVE=
103+
# ENABLE_BULLMQ_DASHBOARD=

‎packages/server/src/CachePool.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,12 @@ export class CachePool {
1212
constructor() {
1313
if (process.env.MODE === MODE.QUEUE) {
1414
if (process.env.REDIS_URL) {
15-
this.redisClient = new Redis(process.env.REDIS_URL)
15+
this.redisClient = new Redis(process.env.REDIS_URL, {
16+
keepAlive:
17+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
18+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
19+
: undefined
20+
})
1621
} else {
1722
this.redisClient = new Redis({
1823
host: process.env.REDIS_HOST || 'localhost',
@@ -26,6 +31,10 @@ export class CachePool {
2631
key: process.env.REDIS_KEY ? Buffer.from(process.env.REDIS_KEY, 'base64') : undefined,
2732
ca: process.env.REDIS_CA ? Buffer.from(process.env.REDIS_CA, 'base64') : undefined
2833
}
34+
: undefined,
35+
keepAlive:
36+
process.env.REDIS_KEEP_ALIVE && !isNaN(parseInt(process.env.REDIS_KEEP_ALIVE, 10))
37+
? parseInt(process.env.REDIS_KEEP_ALIVE, 10)
2938
: undefined
3039
})
3140
}

‎packages/server/src/commands/base.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ export abstract class BaseCommand extends Command {
7676
REDIS_CERT: Flags.string(),
7777
REDIS_KEY: Flags.string(),
7878
REDIS_CA: Flags.string(),
79+
REDIS_KEEP_ALIVE: Flags.string(),
7980
ENABLE_BULLMQ_DASHBOARD: Flags.string()
8081
}
8182

@@ -211,6 +212,7 @@ export abstract class BaseCommand extends Command {
211212
if (flags.QUEUE_REDIS_EVENT_STREAM_MAX_LEN) process.env.QUEUE_REDIS_EVENT_STREAM_MAX_LEN = flags.QUEUE_REDIS_EVENT_STREAM_MAX_LEN
212213
if (flags.REMOVE_ON_AGE) process.env.REMOVE_ON_AGE = flags.REMOVE_ON_AGE
213214
if (flags.REMOVE_ON_COUNT) process.env.REMOVE_ON_COUNT = flags.REMOVE_ON_COUNT
215+
if (flags.REDIS_KEEP_ALIVE) process.env.REDIS_KEEP_ALIVE = flags.REDIS_KEEP_ALIVE
214216
if (flags.ENABLE_BULLMQ_DASHBOARD) process.env.ENABLE_BULLMQ_DASHBOARD = flags.ENABLE_BULLMQ_DASHBOARD
215217
}
216218
}

0 commit comments

Comments
 (0)