LaunchDarkly across multiple celery tasks

We were running into an issue recently where LaunchDarkly wasn't evaluating on celery servers. Opening a python shell on the boxes showed the keys were setup correctly, and that if we called the methods directly, the flags evaluated as expected. However, when called as a delayed task, the flags weren't evaluating.

LaunchDarkly makes use of fork in python, and requires that only one LaunchDarkly client instance exists. This post is a good primer on forking in python. It appeared that this was the issue.

My coworker Doug explained it thusly:

The LaunchDarkly library makes use of threading, Celery starts the main "control" process that uses fork() to start n workers, based on your concurrency setting.

Forking copies the current process into a new one, but in Python it kills off all but the thread doing the forking. All others are stopped. So the threads that LaunchDarkly starts up during initialization (e.g., EventDispatcher or StreamingUpdateProcessor) end up "defunct" or unpredictable.

The Locks used within LaunchDarkly are thread independent, but because threads are killed off in the child, you end up with an invalid state and can’t trust things will work.

Further, LaunchDarkly recommends a post fork hook to initialize the client.

import uwsgidecorators

@uwsgidecorators.postfork
def post_fork_client_initialization():
    ldclient.set_config(LDConfig("sdk-key-123abc"))
    client = ldclient.get()
end

However, our Django application uses asgi, which doesn't currently have this hook. This is our current LaunchDarkly configuration launch_darkly.py:

import atexit
import sys

from django.conf import settings

"""
Sets up Launch Darkly for use across the site.
LD is already initialized. See discovery_service.views.ld_check for example usage.
"""


class LDClient():
    def __getattr__(self, v):
        if 'ldclient' in sys.modules:
            import ldclient
        else:
            import ldclient
            from ldclient.config import Config
            ldclient.set_config(Config(settings.LAUNCH_DARKLY_SDK_KEY))
        return getattr(ldclient.get(), v)


ld_client = LDClient()

@atexit.register
def close_ld(*args, **kwargs):
    # LD recommends closing upon app shutdown
    # https://docs.launchdarkly.com/sdk/server-side/python
    ld_client.close()

The LDClient class allows us to ignore new instantiations of the ldclient library if it's already been loaded.

And the general use is:

from launch_darkly import ld_client

def flagged_code():
    flag = ld_client.variation("flag-name", {"key": 12345}, False)  # False is the default in this case
    if flag:
        // do something if the flag is on
    else:
        // do something else if the flag is off

After a lot of bashing through walls, aka iterative development, we discovered two things:

1. Module Level instantiation

There was a module-level instantiation of the LaunchDarkly client that was causing the library to initialize before the fork.

Basically, the above code, but instead:

from launch_darkly import ld_client

flag = ld_client.variation("flag-name", {"key": 12345}, False)      # False is the default in this case


def flagged_code():

    if flag:
        // do something if the flag is on
    else:
        // do something else if the flag is off

So that code was removed / refactored.

2. Celery initialization

In our celery.py code, we added a worker_process_init hook to initialize the library properly. This ensures that when the celery workers fork, there is definitely a ldclient ready to go for any code that requires it.

@worker_process_init.connect
def configure_worker(signal=None, sender=None, **kwargs):
    """Initialize the Launch Darkly client for use in Celery tasks."""
    try:
        res = ld_client.variation("test-flag", {"key": 0}, 0)
        logging.info(f"LD client initialized for Celery worker. {res}")
    except Exception:
        import traceback
        traceback.print_exc()
        logger.error("Error initializing LD client for Celery worker.", exc_info=True)

To aid in future discovery and debugging, we also created a celery task that we can call on the fly to make sure things are working:

@shared_task
def celery_ld_check(flag="test-flag", key=0, default="not found"):
    """
    Test LaunchDarkly SDK connectivity from Celery.
    """

    print("trying celery_ld_check")
    try:
        variation = ld_client.variation(flag, {"key": key}, default)
        print(f"celery_ld_check: {variation}")
    except Exception as e:
        print(f"celery_ld_check: {e}")

Lastly, we will likely iterate on the LDClient class to deal with issues regarding the fork on the fly.

Let me know if this helps you in your code, or sparks any ideas for you!