Pull request update/241104

arcee/Dockerfile

@@ -6,6 +6,7 @@ ENV PYTHONPATH /usr/src/app/
 
 COPY arcee/arcee_receiver/requirements.txt arcee/arcee_receiver/requirements.txt
 COPY optscale_client/aconfig_cl optscale_client/aconfig_cl
+COPY optscale_client/config_client optscale_client/config_client
 COPY tools/optscale_time tools/optscale_time
 
 RUN pip install --no-cache-dir -r /usr/src/app/arcee/arcee_receiver/requirements.txt

arcee/arcee_receiver/models.py

@@ -8,7 +8,7 @@
 from typing_extensions import Annotated
 
 
-class ArceeState:
+class ArceeState(int, Enum):
     STARTED = 1
     FINISHED = 2
     ERROR = 3
@@ -115,6 +115,17 @@ class TaskPatchIn(BaseClass):
     owner_id: Optional[str] = None
 
 
+class TaskPostIn(TaskPatchIn):
+    key: str
+    metrics: List[str] = []
+
+
+class Task(TaskPostIn):
+    id: str = id_
+    token: str
+    deleted_at: int = 0
+
+
 class LeaderboardFilter(BaseClass):
     id: str = Field(description='metric id to filter by')
     min: Optional[float] = None

arcee/arcee_receiver/producer.py

@@ -0,0 +1,59 @@
+import os
+import asyncio
+from functools import partial
+import concurrent.futures
+
+from kombu import Connection as QConnection, Exchange
+from kombu.pools import producers
+
+
+from optscale_client.config_client.client import Client as ConfigClient
+
+
+class ActivitiesTaskProducer:
+    EXCHANGE_NAME = 'activities-tasks'
+    RETRY_POLICY = {'max_retries': 15, 'interval_start': 0,
+                    'interval_step': 1, 'interval_max': 3}
+    RESCHEDULE_TIMEOUT = 60 * 60 * 12
+
+    executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)
+
+    def __init__(self):
+        self._config_cl = None
+
+    @classmethod
+    async def run_async(cls, func, *args, loop=None, executor=None, **kwargs):
+        if loop is None:
+            loop = asyncio.get_event_loop()
+        if executor is None:
+            executor = cls.executor
+        pfunc = partial(func, *args, **kwargs)
+        return await loop.run_in_executor(executor, pfunc)
+
+    @property
+    def config_cl(self):
+        if self._config_cl is None:
+            config_cl = ConfigClient(
+                host=os.environ.get('HX_ETCD_HOST'),
+                port=int(os.environ.get('HX_ETCD_PORT')),
+            )
+            self._config_cl = config_cl
+        return self._config_cl
+
+    def create_task(self, task, routing_key):
+        params = self.config_cl.read_branch('/rabbit')
+        conn_str = f'amqp://{params["user"]}:{params["pass"]}@' \
+                   f'{params["host"]}:{params["port"]}'
+        queue_conn = QConnection(conn_str, transport_options=self.RETRY_POLICY)
+
+        task_exchange = Exchange(self.EXCHANGE_NAME, type='topic')
+        with producers[queue_conn].acquire(block=True) as producer:
+            producer.publish(
+                task,
+                serializer='json',
+                exchange=task_exchange,
+                declare=[task_exchange],
+                routing_key=routing_key,
+                retry=True,
+                retry_policy=self.RETRY_POLICY
+            )

arcee/arcee_receiver/requirements.txt

@@ -2,10 +2,12 @@ aiohttp==3.10.2
 sanic==23.12.1
 sanic-ext==23.12.0
 motor==3.6.0
+kombu==5.3.4
 pymongo==4.9.1
 python-etcd==0.4.5
 mongodb-migrations==1.2.1
 pydantic==2.4.2
 # OptScale packages
+-e optscale_client/config_client
 -e optscale_client/aconfig_cl
 -e tools/optscale_time