Skip to content

🐛 fix rabbit mq unique queue name #6328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

🐛 fix rabbit mq unique queue name #6328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion packages/service-library/src/servicelib/rabbitmq/_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import logging
import random
import socket
import string
from typing import Any, Final

import aio_pika
Expand All @@ -19,6 +21,9 @@

RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS: Final[int] = 15 * _MINUTE * 1000

CHARACTERS = string.ascii_letters + string.digits
_GENERATE_RANDOM_STRING_LENGTH = 6
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@pcrespov I guess I can create a function for this and add it to utils_secrets.py? we do not yet have one that combines letters and numbers as far as I can see.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess here generate_passcode might work for this case as well.

In any case, more than a secure here the idea is to generate some sort of unique identifer.
I would start a new module with helpesr like servicelib.identifiers_utils and add functions that can produce unique identifiers e.g. based on some unique feature (time, hostname, etc...)

import hashlib
import time
from models_library.basic_types import IdStr

def short_sha256(input_string, length=8) -> IdStr:
    sha_signature = hashlib.sha256(input_string.encode()).hexdigest()
    return IdStr(sha_signature[:length])

# Example usage
def get_rabbitma_client_unique_name(prefix: str)
   hostname = socket.gethostname()
   unique_id = short_sha256( f"{time.time()}" + hostname , length=8)
   return f"{prefix}_{hostname}_{unique_id}"

NOTE that the value you pass to input_string provides the "uniqueness" scope of that value. Therefore you use as discriminator.



class RabbitMQRetryPolicyUponInitialization:
"""Retry policy upon service initialization"""
Expand Down Expand Up @@ -51,7 +56,10 @@ async def wait_till_rabbitmq_responsive(url: str) -> bool:


def get_rabbitmq_client_unique_name(base_name: str) -> str:
return f"{base_name}_{socket.gethostname()}"
random_string = "".join(
random.choice(CHARACTERS) for _ in range(_GENERATE_RANDOM_STRING_LENGTH)
)
return f"{base_name}_{socket.gethostname()}_{random_string}"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not convinced by this. The whole idea of using the hostname here was to make it unique.

I would rather drop the hostname and use a larger random string, which makes collisions less likely. You might even get away with using a uuid from which you strip the -.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please note that all your tests will fail if something fishy is happening with the queue names. You might want to check what your change introduced.



async def declare_queue(
Expand Down
6 changes: 3 additions & 3 deletions ...server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@

_logger = logging.getLogger(__name__)

_APP_RABBITMQ_CONSUMERS_KEY: Final[str] = f"{__name__}.rabbit_consumers"
_APP_RABBITMQ_EXCLUSIVE_CONSUMERS_KEY: Final[str] = f"{__name__}.rabbit_consumers"


async def _convert_to_node_update_event(
Expand Down Expand Up @@ -181,15 +181,15 @@ async def _unsubscribe_from_rabbitmq(app) -> None:
await logged_gather(
*(
rabbit_client.unsubscribe(queue_name)
for queue_name in app[_APP_RABBITMQ_CONSUMERS_KEY].values()
for queue_name in app[_APP_RABBITMQ_EXCLUSIVE_CONSUMERS_KEY].values()
),
)


async def on_cleanup_ctx_rabbitmq_consumers(
app: web.Application,
) -> AsyncIterator[None]:
app[_APP_RABBITMQ_CONSUMERS_KEY] = await subscribe_to_rabbitmq(
app[_APP_RABBITMQ_EXCLUSIVE_CONSUMERS_KEY] = await subscribe_to_rabbitmq(
app, _EXCHANGE_TO_PARSER_CONFIG
)
yield
Expand Down