Repository: Unleash/unleash-client-python (python)

Issue:
Unleash apscheduler executor should not use `"default"` executor
**Is your feature request related to a problem? Please describe.**
We're using apscheduler in our project for 1.5 year now. We added unleash few months ago. Recently, we started to observe weird process/thread locking within apscheduler jobs. Today we looked into our and Unleash code and it turned out that we both were using the same executor. We had our own defined as 
```python
executors = {
   "default": ThreadPoolExecutor(1),
}
```
which - I believe, because I don't have skills to investigate that, but I've seen logs that could suggest that - clashes with the default executor that Unleash client is using. 


**Describe the solution you'd like**
- allow developers to change executor name, but use own executor with name partially random if no name is specified in initializer

**Describe alternatives you've considered**
- Unleash should create it's own executor with name partially random to prevent executor names clashing, e.i. `"unleash_executor_6f5a3n"`
- force developers to name the executor explicitly and let them know that there is apscheduler underneath

**Additional context**
Of course we made a mistake of using the default executor. We should've started with our own unique name. But that shows the behavior that may blow up someone else's application too. 

Suppose, they're using apscheduler for years. Recently, they decided to add a third party application that uses apscheduler which crashes the app. Now, the team has to make changes to the working app, so that it would accommodate a new third party. 

To me, this looks like giving up a drivers' seat to a hitchhiker, rather than inviting for a free ride on the drivers' rules. Third party should have all the set up needed for being self sufficient without a need to bother the host.


Pull request description:
Allow customization of  APScheduler
# Description

Fixes #221

* Allows users to pass in a custom APScheduler `scheduler` and specify the executor type.
* If no customized `scheduler` is provided, spin up an scheduler with a Unleash-specific pool name.

## Type of change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

# How Has This Been Tested?

Please describe the tests that you ran to verify your changes.

- [ ] Unit tests
- [ ] Spec Tests
- [ ] Integration tests / Manual Tests

# Checklist:

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my own code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream modules

Patch:
diff --git a/UnleashClient/__init__.py b/UnleashClient/__init__.py
index 170f018..81e91eb 100644
--- a/UnleashClient/__init__.py
+++ b/UnleashClient/__init__.py
@@ -1,10 +1,14 @@
 # pylint: disable=invalid-name
 import warnings
+import random
+import string
 from datetime import datetime, timezone
 from typing import Callable, Optional
 from apscheduler.job import Job
+from apscheduler.schedulers.base import BaseScheduler
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
+from apscheduler.executors.pool import ThreadPoolExecutor
 from UnleashClient.api import register_client
 from UnleashClient.periodic_tasks import fetch_and_load_features, aggregate_and_send_metrics
 from UnleashClient.strategies import ApplicationHostname, Default, GradualRolloutRandom, \
@@ -35,6 +39,8 @@ class UnleashClient:
     :param cache_directory: Location of the cache directory. When unset, FCache will determine the location.
     :param verbose_log_level: Numerical log level (https://docs.python.org/3/library/logging.html#logging-levels) for cases where checking a feature flag fails.
     :param cache: Custom cache implementation that extends UnleashClient.cache.BaseCache.  When unset, UnleashClient will use Fcache.
+    :param scheduler: Custom APScheduler object.  Use this if you want to customize jobstore or executors.  When unset, UnleashClient will create it's own scheduler.
+    :param scheduler_executor: Name of APSCheduler executor to use if using a custom scheduler.
     """
     def __init__(self,
                  url: str,
@@ -53,7 +59,9 @@ class UnleashClient:
                  cache_directory: Optional[str] = None,
                  project_name: str = None,
                  verbose_log_level: int = 30,
-                 cache: Optional[BaseCache] = None) -> None:
+                 cache: Optional[BaseCache] = None,
+                 scheduler: Optional[BaseScheduler] = None,
+                 scheduler_executor: Optional[str] = None) -> None:
         custom_headers = custom_headers or {}
         custom_options = custom_options or {}
         custom_strategies = custom_strategies or {}
@@ -80,7 +88,6 @@ class UnleashClient:
 
         # Class objects
         self.features: dict = {}
-        self.scheduler = BackgroundScheduler()
         self.fl_job: Job = None
         self.metric_job: Job = None
 
@@ -91,6 +98,27 @@ class UnleashClient:
         })
         self.unleash_bootstrapped = self.cache.bootstrapped
 
+        # Scheduler bootstrapping
+        # - Figure out the Unleash executor name.
+        if scheduler and scheduler_executor:
+            self.unleash_executor_name = scheduler_executor
+        elif scheduler and not scheduler_executor:
+            raise ValueError("If using a custom scheduler, you must specify a executor.")
+        else:
+            if not scheduler:
+                LOGGER.warning("scheduler_executor should only be used with a custom scheduler.")
+
+            self.unleash_executor_name = f"unleash_executor_{''.join(random.choices(string.ascii_uppercase + string.digits, k=6))}"
+
+        # Set up the scheduler.
+        if scheduler:
+            self.unleash_scheduler = scheduler
+        else:
+            executors = {
+                self.unleash_executor_name: ThreadPoolExecutor()
+            }
+            self.unleash_scheduler = BackgroundScheduler(executors=executors)
+
         # Mappings
         default_strategy_mapping = {
             "applicationHostname": ApplicationHostname,
@@ -184,20 +212,22 @@ class UnleashClient:
 
                 job_func(**job_args)  # type: ignore
                 # Start periodic jobs
-                self.scheduler.start()
-                self.fl_job = self.scheduler.add_job(job_func,
+                self.unleash_scheduler.start()
+                self.fl_job = self.unleash_scheduler.add_job(job_func,
                                                      trigger=IntervalTrigger(
                                                          seconds=int(self.unleash_refresh_interval),
                                                          jitter=self.unleash_refresh_jitter,
                                                      ),
+                                                     executor=self.unleash_executor_name,
                                                      kwargs=job_args)
 
                 if not self.unleash_disable_metrics:
-                    self.metric_job = self.scheduler.add_job(aggregate_and_send_metrics,
+                    self.metric_job = self.unleash_scheduler.add_job(aggregate_and_send_metrics,
                                                              trigger=IntervalTrigger(
                                                                  seconds=int(self.unleash_metrics_interval),
                                                                  jitter=self.unleash_metrics_jitter,
                                                              ),
+                                                             executor=self.unleash_executor_name,
                                                              kwargs=metrics_args)
             except Exception as excep:
                 # Log exceptions during initialization.  is_initialized will remain false.
@@ -218,7 +248,7 @@ class UnleashClient:
         self.fl_job.remove()
         if self.metric_job:
             self.metric_job.remove()
-        self.scheduler.shutdown()
+        self.unleash_scheduler.shutdown()
         self.cache.destroy()
 
     @staticmethod
diff --git a/pyproject.toml b/pyproject.toml
index 65a16b8..6582c21 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -28,7 +28,7 @@ disable = [
 ]
 max-attributes = 25
 max-args = 25
-max-locals = 20
+max-locals = 25
 extension-pkg-allow-list = ["mmh3"]
 
 [tool.setuptools_scm]


Test patch:
diff --git a/tests/unit_tests/test_client.py b/tests/unit_tests/test_client.py
index 9ba9568..e676359 100644
--- a/tests/unit_tests/test_client.py
+++ b/tests/unit_tests/test_client.py
@@ -5,6 +5,8 @@ from pathlib import Path
 
 import pytest
 import responses
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.executors.pool import ThreadPoolExecutor
 from UnleashClient import UnleashClient
 from UnleashClient.strategies import Strategy
 from tests.utilities.testing_constants import URL, ENVIRONMENT, APP_NAME, INSTANCE_ID, REFRESH_INTERVAL, REFRESH_JITTER, \
@@ -228,7 +230,7 @@ def test_uc_dirty_cache(unleash_client_nodestroy):
     unleash_client.initialize_client()
     time.sleep(5)
     assert unleash_client.is_enabled("testFlag")
-    unleash_client.scheduler.shutdown()
+    unleash_client.unleash_scheduler.shutdown()
 
     # Check that everything works if previous cache exists.
     unleash_client.initialize_client()
@@ -492,3 +494,44 @@ def test_uc_cache_bootstrap_url(cache):
     )
     assert len(unleash_client.features) >= 4
     assert unleash_client.is_enabled("testFlag")
+
+
+@responses.activate
+def test_uc_custom_scheduler():
+    # Set up API
+    responses.add(responses.POST, URL + REGISTER_URL, json={}, status=202)
+    responses.add(responses.GET, URL + FEATURES_URL, json=MOCK_FEATURE_RESPONSE, status=200, headers={'etag': ETAG_VALUE})
+    responses.add(responses.POST, URL + METRICS_URL, json={}, status=202)
+
+    # Set up UnleashClient
+    custom_executors = {
+        'hamster_executor': ThreadPoolExecutor()
+    }
+
+    custom_scheduler = BackgroundScheduler(
+        executors=custom_executors
+    )
+
+    unleash_client = UnleashClient(
+        URL,
+        APP_NAME,
+        refresh_interval=5,
+        metrics_interval=10,
+        scheduler=custom_scheduler,
+        scheduler_executor='hamster_executor'
+    )
+
+    # Create Unleash client and check initial load
+    unleash_client.initialize_client()
+    time.sleep(1)
+    assert unleash_client.is_initialized
+    assert len(unleash_client.features) >= 4
+
+    # Simulate caching
+    responses.add(responses.GET, URL + FEATURES_URL, json={}, status=304, headers={'etag': ETAG_VALUE})
+    time.sleep(6)
+
+    # Simulate server provisioning change
+    responses.add(responses.GET, URL + FEATURES_URL, json=MOCK_ALL_FEATURES, status=200, headers={'etag': 'W/somethingelse'})
+    time.sleep(6)
+    assert len(unleash_client.features) >= 9