support redis cluster transport

[zs-neo]

Attempt to address #1021

Thank you very much for your code, it helps us a lot. @auvipy

We use redis-py instead of redis-py-cluster because redis-py-cluster has been merged into redis-py.
Celery works fine on our cluster with multi producers and multi consumers, when a node goes down, it can automatically switch.

celery==5.4.0
Django==4.1
django-celery-beat==2.7.0
django-celery-results==2.5.1
django-filter==24.3
django-redis==5.4.0
django-split-settings==0.3.0
django-timezone-field==5.1
djangorestframework==3.15.2
gunicorn==22.0.0
pyOpenSSL==21.0.0
redis==5.0.0
requests==2.22.0
uWSGI==2.0.18

app = Celery('celery_test', broker='rediscluster://:myredis@127.0.0.1:7024')

[codecov]

Codecov Report

Attention: Patch coverage is 86.76123% with 56 lines in your changes missing coverage. Please review.

Project coverage is 81.89%. Comparing base (13e6938) to head (543cc93).
Report is 12 commits behind head on main.

Files with missing lines	Patch %	Lines
kombu/transport/rediscluster.py	86.76%	34 Missing and 22 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2204      +/-   ##
==========================================
+ Coverage   81.60%   81.89%   +0.28%     
==========================================
  Files          77       78       +1     
  Lines        9540     9963     +423     
  Branches     1162     1238      +76     
==========================================
+ Hits         7785     8159     +374     
- Misses       1563     1591      +28     
- Partials      192      213      +21     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[auvipy]

lets see if the CI passes

[auvipy]

Only one test is failing, due to connection failure

[zs-neo]

@auvipy @Nusnus ping :)
I fixed the issues I found recently and now this PR is ready.
I'm not a kombu expert, any suggestions are welcome

[Nusnus]

Check linter error :)

[zs-neo]

Sorry for the failure, I will fix it and improve the test coverage.

[auvipy]

I am following it and already had reviewed it twice. Will have an in depth review again tomorrow. No worries. Thanks for picking my work

[Nusnus]

https://github.com/celery/kombu/actions/runs/12492830091/job/34860983164?pr=2204

[zs-neo]

https://github.com/celery/kombu/actions/runs/12492830091/job/34860983164?pr=2204

Added docs for kombu.transport.rediscluster, now I'm sure there won't be problems anymore

[Nusnus]

Finally that everything passes on the CI, we can properly review the code.
That being said, we’re currently in a release phase so we can’t merge it until we complete the release, after the new year holidays.

Good work and thank you for fixing everything.

[lan17]

Would this also support Redis cluster for backend?

[jnu]

@zs-neo thanks for all of your work on this! I'm testing out this PR right now and noticed that TLS support is broken. it looks as though it might be as simple as using the rediss scheme here with the ssl logic above?

[zs-neo]

I'm very sorry for the late reply. Thank you for your attention to this PR! I agree with you and I will add support and testing for TLS.

[auvipy]

did you add the test for TLS?

[rfenner]

It would be nice to see this land sooon as I'm trying to use AWS serverless valkey and it runs in a cluster mode so Celery won't run on it. It's going to become more common for people to want to run their projects on serverless versions since they tend to offer scalable storage and not a fixed amount like you would get when you use a specific instance type.

I guess my workaround for now will be to spin up a normal version just for celery to use.

[Copilot]

Pull Request Overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

kombu/transport/rediscluster.py:1

• Typo in variable name: 'disconect' should be 'disconnect'

"""Redis cluster transport module for Kombu.

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Ragnarow]

Hello team, any update on this one? Looks like it's very close to get shipped to next release?

[Ragnarow]

@auvipy Looks like it's a documentation issue, is it transient, or is there something else needed?

[auvipy]

this isslated for v5.7.0 release

[auvipy]

=================================== FAILURES ===================================
_____________________ test_Channel.test_prefixed_pipeline ______________________

self = <t.unit.transport.test_rediscluster.test_Channel object at 0x7f074f0af7c0>
mock_initialize =
mock_execute_command =

@patch("redis.cluster.RedisCluster.execute_command")
@patch('redis.cluster.NodesManager.initialize')
def test_prefixed_pipeline(self, mock_initialize, mock_execute_command):
    client = redis.PrefixedStrictRedis(global_keyprefix='foo_', startup_nodes=[ClusterNode()])

  pipeline = client.pipeline()

t/unit/transport/test_rediscluster.py:614:

---

self = <kombu.transport.rediscluster.PrefixedStrictRedis object at 0x7f0736200ca0>
transaction = False, shard_hint = None

Restoring 2 unacknowledged message(s)
def pipeline(self, transaction=False, shard_hint=None):
if shard_hint:
raise RedisClusterException("shard_hint is deprecated in cluster mode")
if transaction:
raise RedisClusterException("transaction is deprecated in cluster mode")

    return PrefixedRedisPipeline(
        nodes_manager=self.nodes_manager,
        commands_parser=self.commands_parser,
        startup_nodes=self.nodes_manager.startup_nodes,
        result_callbacks=self.result_callbacks,
        cluster_response_callbacks=self.cluster_response_callbacks,

      cluster_error_retry_attempts=self.cluster_error_retry_attempts,

        read_from_replicas=self.read_from_replicas,
        reinitialize_steps=self.reinitialize_steps,
        lock=self._lock,
        global_keyprefix=self.global_keyprefix,
    )

E AttributeError: 'PrefixedStrictRedis' object has no attribute 'cluster_error_retry_attempts'

kombu/transport/rediscluster.py:133: AttributeError
=============================== warnings summary ===============================
.tox/3.9-unit/lib/python3.9/site-packages/coverage/inorout.py:473
/home/runner/_work/kombu/kombu/.tox/3.9-unit/lib/python3.9/site-packages/coverage/inorout.py:473: CoverageWarning: --include is ignored because --source is set (include-ignored); see https://coverage.readthedocs.io/en/7.10.7/messages.html#warning-include-ignored
self.warn("--include is ignored because --source is set", slug="include-ignored")

.tox/3.9-unit/lib/python3.9/site-packages/google/api_core/_python_version_support.py:252
/home/runner/_work/kombu/kombu/.tox/3.9-unit/lib/python3.9/site-packages/google/api_core/_python_version_support.py:252: FutureWarning: You are using a Python version (3.9.23) past its end of life. Google will update google.api_core with critical bug fixes on a best-effort basis, but not with any other fixes or features. Please upgrade to the latest Python version, or at least Python 3.10, and then update google.api_core.
warnings.warn(message, FutureWarning)

[phmcder]

Are there any updates on this?

[auvipy]

no, probably need a new taker. @MehrazRumman can you take this please? and come with a new PR on a new branch?