Bull: Redis master/slave setup on Kubernetes throwing error: BRPOPLPUSH { ReplyError: MOVED 2651

[hagginoaks]

Description

I'm trying to debug this issue I'm seeing on Kubernetes where I have redis configured as a cluster.

When Kubernetes restarts upon deployments, I run into this following error:

BRPOPLPUSH { ReplyError: MOVED 2651 <IP_ADDRESS>:6379
at parseError (/usr/src/app/node_modules/ioredis/node_modules/redis-parser/lib/parser.js:179:12)
at parseType (/usr/src/app/node_modules/ioredis/node_modules/redis-parser/lib/parser.js:302:14)
command:
{ name: 'brpoplpush',
args:
[ '{slack}:slack notifications:wait',
'{slack}:slack notifications:active',
'5' ] } }

Where is, I think the cluster IP? I didn't configure this, but I'm trying to debug this. I want to know if I need to enable cluster mode for Bull.js or if this is a configuration issue outside of the Bull.js project?

Or is it a networking issue with K8s?

Would enabling: https://github.com/OptimalBits/bull#cluster-support be the solution? Is this the right approach?

Here is my code:

Minimal, Working Test code to reproduce the issue.

(An easy to reproduce test case will dramatically decrease the resolution time.)

This is where I initialize the queues.

import Queue from 'bull';
import config from 'config';
import { run as slackRun } from './tasks/send-slack-message';
import { run as emailRun } from './tasks/send-email';

const redisConfig = {
  redis: {
    host: config.redis.host,
    port: config.redis.port
  }
};

const slackQueue = new Queue('slack notifications', { ...redisConfig, ...{ prefix: '{slack}' } });
const emailQueue = new Queue('email notifications', { ...redisConfig, ...{ prefix: '{email}' } });

slackQueue.process(slackRun);
emailQueue.process(emailRun);

emailQueue.on('completed', (job, result) => {
  job.remove();
});

export { emailQueue, slackQueue };

Then when I add a job to the queue after a specific action:

import { emailQueue, slackQueue } from 'worker/worker';

const queueOptions = {
  attempts: 2,
  removeOnComplete: true,
  backoff: {
    type: 'exponential',
    delay: 60 * 1000
  }
};

emailQueue.add(
  {
    params: {
      from: email,
      fromname: name,
      text: body
    }
  },
  queueOptions
);
slackQueue.add(
  {
    channelId: SLACK_CHANNELS.FEEDBACK,
    attachments: [
      {
        text: req.body.body
      }
    ]
  },
  queueOptions
);

Bull version

3.6.0

Additional information

Here is the configmap:

Name:         redis-cluster-config
Namespace:    default
Labels:       <none>
Annotations:  <none>

Data
====
update-node.sh:
----
#!/bin/sh
REDIS_NODES="/data/nodes.conf"
sed -i -e "/myself/ s/[0-9]\{1,3\}\.[0-9]\{1,3\}\.[0-9]\{1,3\}\.[0-9]\{1,3\}/${POD_IP}/" ${REDIS_NODES}
exec "$@"

redis.conf:
----
cluster-enabled yes
cluster-require-full-coverage no
cluster-node-timeout 15000
cluster-config-file nodes.conf
cluster-migration-barrier 1
appendonly yes
# Other cluster members need to be able to connect
protected-mode no

Events:  <none>

[manast]

ok, I do not know why this error appears, I would need to setup a cluster myself to debug it, it is probably an issue with ioredis instead of bull anyway.
Having said that, you will not gain so much using a cluster setup for Bull, so unless you are using the cluster for other stuff, a simple master->slave redis setup (slave for redundance), would be more than enough.

[hagginoaks]

@manast thanks for the quick response. :)

When setting up a simple master->slave redis setup, is there anything special I need to do when configuring a Bull queue?

[manast]

nope, just use the IP or address of the master.

[jlvallelonga]

Has anyone else seen this issue? I'm getting this same error now connecting to a cluster using the nestjs wrapper. Is there a solution other than just using a simple master-> slave setup?

[omeraplak]

@jlvallelonga I have same issue with nestjs. Did you find a solution?

[tudorconstantin]

I managed to get rid of those bull BRPOPLPUSH ReplyError: MOVED errors by building an ioredis connection and reusing it:

// redisOpts contains the usual redis connect params, plus redisOpts.clustered which I use to differentiate between cluster and non-cluster mode. I'm sure there must be other methods for doing this, but I was in a hurry to find a fix
const _getConnOpts = function (redisOpts) {
    // these are for non cluster redis mode
    if (!redisOpts?.clustered) {
      return {
        redis: {
          host: redisOpts?.redis?.host || 'localhost',
          port: redisOpts?.redis?.port || 6379,
          password: redisOpts?.redis?.password || null,
        },
        prefix: redisOpts?.redis?.prefix || '{worker}',
      }
    }

    const Redis = require('ioredis');
    const client = new Redis.cluster([{ host: redisOpts.host }], { dnsLookup: (address, callback) => callback(null, address) });
    const subscriber = new Redis.cluster([{ host: redisOpts.host }], { dnsLookup: (address, callback) => callback(null, address) });
    let opts = {
      createClient: function (type) {
        switch (type) {
          case 'client':
            return client;
          case 'subscriber':
            return subscriber;
          default:
            return new Redis.cluster([{ host: redisOpts.host }], { dnsLookup: (address, callback) => callback(null, address) });
        }
      }
    }

And then, instantiate your queue like const myQueue = new Queue('myQueue', __getConnOpts(redisOpts))

Hope this helps!