resource monitor

Author: Reapor-Yurnero

projects/sbos-minimal/sbos/minimal/interfaces/actuation/actuation_interface.py

@@ -45,6 +45,7 @@ async def actuate(self, domain, entity_id, value):
             )
             actuation_time = time.time() - start  # for benchmark
         except Exception as e:
+            logger.exception(e)
             success, detail = False, f"{e}"
         return success, detail, driver_time, actuation_time
 

projects/sbos-minimal/sbos/minimal/interfaces/actuation/metasys/__init__.py

@@ -7,14 +7,14 @@
 
 class MetasysActuation(BaseActuation):
     def __init__(self, *args, **kwargs):
-        pass
+        self.channel_address = "derconnect-brick-bm.sdsc.edu:50051"
 
     async def actuate(self, entity_id, value, external_references):
         sensor_id = external_references[
             "https://brickschema.org/schema/Brick/ref#metasysID"
         ]
-        logger.info("metasys: {} {}", sensor_id, value)
-        async with grpc.aio.insecure_channel("172.17.0.1:50051") as channel:
+        logger.info("metasys actuate: {} -> {} {}", entity_id, sensor_id, value)
+        async with grpc.aio.insecure_channel(self.channel_address) as channel:
             stub = actuate_pb2_grpc.ActuateStub(channel)
             response: actuate_pb2.Response = await stub.TemporaryOverride(
                 actuate_pb2.TemporaryOverrideAction(
@@ -28,8 +28,8 @@ async def read(self, entity_id, external_references):
         sensor_id = external_references[
             "https://brickschema.org/schema/Brick/ref#metasysID"
         ]
-        logger.info("metasys read: {}", sensor_id)
-        async with grpc.aio.insecure_channel("172.17.0.1:50051") as channel:
+        logger.info("metasys read: {} -> {}", entity_id, sensor_id)
+        async with grpc.aio.insecure_channel(self.channel_address) as channel:
             stub = actuate_pb2_grpc.ActuateStub(channel)
             response: actuate_pb2.Response = await stub.ReadObjectCurrent(
                 actuate_pb2.ReadObjectCurrentAction(

projects/sbos-minimal/sbos/minimal/interfaces/timeseries/asyncpg/__init__.py

@@ -404,6 +404,7 @@ async def add_history_data(
     ):
         table_name = self.get_history_table_name(domain_name)
         async with self.pool.acquire() as conn:
+            logger.info("{}", domain_name)
             res = await conn.execute(
                 f"""INSERT INTO {table_name} (uuid, user_id, app_name, domain_user_app, time, value)
                 VALUES ('{entity_id}', '{user_id}', '{app_name}', '{domain_user_app}', '{time}', '{value}');"""

projects/sbos-minimal/sbos/minimal/services/actuation.py

@@ -5,6 +5,7 @@
 from fastapi import APIRouter, Body, Depends
 from fastapi_restful.cbv import cbv
 from starlette.requests import Request
+from loguru import logger
 
 from sbos.minimal import models, schemas
 from sbos.minimal.interfaces import ActuationInterface, BaseTimeseries, GraphDB
@@ -66,11 +67,12 @@ async def actuate_entity(self, domain, jwt_payload, entity_id, actuation_payload
             ) = await self.actuation_iface.actuate(
                 domain, entity_id, actuation_payload[0]
             )
+            logger.info("success: {}, detail: {}", success, detail)
             await self.ts_db.add_history_data(
                 domain.name,
                 entity_id,
                 jwt_payload["sub"],
-                jwt_payload["app_name"],
+                jwt_payload["app"],
                 jwt_payload.get("domain_user_app", ""),
                 arrow.now(),
                 actuation_payload[0],
@@ -81,6 +83,7 @@ async def actuate_entity(self, domain, jwt_payload, entity_id, actuation_payload
                 (policy_time, guard_time, driver_time, actuation_time),
             )
         except Exception as e:
+            logger.exception(e)
             return False, f"{e}", (policy_time, guard_time, driver_time, actuation_time)
 
     @router.post(

projects/sbos-monitor/sbos/monitor/__main__.py

@@ -51,9 +51,10 @@ async def validate_resources_in_domain(domain, domain_data):
                     violated_constraints[entity_id] = real_value
         else:
             logger.error(response)
+            logger.error(response_dict)
 
         if len(violated_constraints) > 0:
-            url = f"{playground_api_base}/domains/{domain}/resources"
+            url = f"{playground_api_base}/domains/{domain}/resources/notify"
             data = {k: [str(v)] for k, v in violated_constraints.items()}
             response = await client.post(url, json=data, headers=headers)
             response_dict = response.json()

projects/sbos-monitor/sbos/monitor/config/settings/base.py

@@ -24,7 +24,7 @@ class MonitorBaseSettings(DatabaseMongoDBSettings):
     # VERSION: str = "0.1.0"
     PLAYGROUND_HOST: str = Field(default="localhost", description="The hostname of playground server.")
     PLAYGROUND_PORT: int = Field(default=9000, description="The port of playground server")
-    PLAYGROUND_JWT_TOKEN: str = Field(default="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyQGV4YW1wbGUuY29tIiwiYXVkIjpbImJyaWNrIl0sImV4cCI6MTc0NjM4MDAzMX0.LQ8k5qEBTo_6abTNzPdmX3Vkty6iP678Hb2w51NWclo")
+    PLAYGROUND_JWT_TOKEN: str = Field(default="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJsaXV5aDk3MDYxNUBnbWFpbC5jb20iLCJhdWQiOlsiYnJpY2siXSwiZXhwIjoxNzQ2ODI1MzgyfQ.f2p6KSknZ0oWWY769mpZzZWEqDb4HK3h40WhF0Tlj7Q")
     POLLING_INTERVAL: float = Field(default=5, description="The interval of polling in seconds.")
 
     class Config:

projects/sbos-playground/docker-compose.yml

@@ -94,7 +94,7 @@ services:
     environment:
       ME_CONFIG_BASICAUTH_USERNAME: ${MONGO_USERNAME:-root}
       ME_CONFIG_BASICAUTH_PASSWORD: ${MONGO_PASSWORD:-pass}
-      ME_CONFIG_MONGODB_ADMINUSERNAME: ${MONGO_USERNAME:-r  oot}
+      ME_CONFIG_MONGODB_ADMINUSERNAME: ${MONGO_USERNAME:-root}
       ME_CONFIG_MONGODB_ADMINPASSWORD: ${MONGO_PASSWORD:-pass}
 #      ME_CONFIG_MONGODB_URL: mongodb://${MONGO_USERNAME:-root}:${MONGO_PASSWORD:-pass}@mongo:27017/
       ME_CONFIG_MONGODB_URL: mongodb://mongo:27017/

projects/sbos-playground/sbos/playground/interfaces/scheduling_policy/base.py

@@ -6,7 +6,6 @@
 from sbos.minimal.interfaces import ActuationInterface, AsyncpgTimeseries
 
 from sbos.playground import models, schemas
-from sbos.playground.interfaces.app_management import stop_container
 
 
 class SchedulingPolicyBase(ABC):
@@ -30,38 +29,6 @@ async def find_victim(
     ):
         raise NotImplementedError()
 
-    async def kill_app(self, domain_user_app_id):
-        domain_user_app: models.DomainUserApp = await models.DomainUserApp.get(
-            domain_user_app_id
-        )
-        if domain_user_app is not None:
-            logger.info("kill app: {}", domain_user_app)
-            try:
-                stop_container(domain_user_app.container_id)
-            except Exception as e:
-                logger.exception(e)
-
-    async def relinquish_entity(self, entity_id):
-        logger.info("relinquish entity: {}", entity_id)
-        try:
-            await self.actuation_iface.actuate(entity_id, "null")
-        except Exception as e:
-            logger.exception(e)
-
+    @abstractmethod
     async def schedule(self, notify_resource: schemas.ResourceConstraintRead):
-        entity_ids = await self.blame(notify_resource.entity_id)
-        # get history with entity_ids (postgres)
-        history = await self.ts_db.get_history_data(self.domain.name, entity_ids)
-        # [('additionalProp1', 'admin', 'sb', datetime.datetime(2023, 7, 1, 18, 8, 0, 963859)),
-        #  ('additionalProp3', 'admin', 'sb', datetime.datetime(2023, 7, 1, 18, 8, 1, 205563)),
-        #  ('additionalProp1', 'admin', 'sb', datetime.datetime(2023, 7, 2, 2, 37, 25))]
-        logger.info(history)
-        victim_apps, victim_entity_ids = await self.find_victim(
-            history, notify_resource
-        )
-        logger.info("victim apps: {}", victim_apps)
-        logger.info("victim entity_ids: {}", victim_entity_ids)
-        await asyncio.gather(
-            *[self.kill_app(app) for app in victim_apps],
-            *[self.relinquish_entity(entity_id) for entity_id in victim_entity_ids]
-        )
+        raise NotImplementedError()

projects/sbos-playground/sbos/playground/interfaces/scheduling_policy/naive.py

@@ -1,13 +1,17 @@
 import itertools
 from typing import Any, List
 
-from sbos.playground import schemas
+from loguru import logger
+from sbos.minimal.interfaces import ActuationInterface, AsyncpgTimeseries
+
+from sbos.playground import models, schemas
+from sbos.playground.interfaces.app_management import stop_container
 from sbos.playground.interfaces.scheduling_policy.base import SchedulingPolicyBase
 
 
 class SchedulingPolicyNaive(SchedulingPolicyBase):
     async def blame(self, entity_id: str) -> List[str]:
-        return ["additionalProp1", "additionalProp3"]
+        return ["http://ucsd.edu/ontology/building/Center_Hall#AH-6.WC-ADJ"]
 
     async def find_victim(
         self, history: List[Any], notify_resource: schemas.ResourceConstraintRead
@@ -20,3 +24,52 @@ async def find_victim(
         victim_app = app_latest_use_time[-1][3]
         victim_entity = app_latest_use_time[-1][0]
         return [victim_app], [victim_entity]
+
+    async def kill_app(self, domain_user_app_id):
+        domain_user_app: models.DomainUserApp = await models.DomainUserApp.get(
+            domain_user_app_id
+        )
+        if domain_user_app is not None and domain_user_app.status == schemas.DockerStatus.RUNNING:
+            logger.info("kill app: {}", domain_user_app)
+            try:
+                loop = asyncio.get_running_loop()
+                container = await loop.run_in_executor(None, stop_container, domain_user_app.container_id)
+                domain_user_app.status = container.status
+                await domain_user_app.save()
+                return True
+            except Exception as e:
+                logger.exception(e)
+
+        return False
+    async def relinquish_entity(self, entity_id):
+        logger.info("relinquish entity: {}", entity_id)
+        try:
+            await self.actuation_iface.actuate(self.domain, entity_id, "null")
+        except Exception as e:
+            logger.exception(e)
+
+    async def process_app_and_entity(self, domain_user_app_id, entity_id):
+        killed = await self.kill_app(domain_user_app_id)
+        if killed:
+            await self.relinquish_entity(entity_id)
+
+    async def schedule(self, notify_resource: schemas.ResourceConstraintRead):
+        entity_ids = await self.blame(notify_resource.entity_id)
+        # get history with entity_ids (postgres)
+        history = await self.ts_db.get_history_data(self.domain.name, entity_ids)
+        # [('additionalProp1', 'admin', 'sb', datetime.datetime(2023, 7, 1, 18, 8, 0, 963859)),
+        #  ('additionalProp3', 'admin', 'sb', datetime.datetime(2023, 7, 1, 18, 8, 1, 205563)),
+        #  ('additionalProp1', 'admin', 'sb', datetime.datetime(2023, 7, 2, 2, 37, 25))]
+        logger.info(history)
+        victim_apps, victim_entity_ids = await self.find_victim(
+            history, notify_resource
+        )
+        logger.info("victim apps: {}", victim_apps)
+        logger.info("victim entity_ids: {}", victim_entity_ids)
+        if len(victim_apps) > 0 and len(victim_entity_ids) > 0:
+            await self.process_app_and_entity(victim_apps[0], victim_entity_ids[0])
+
+        # await asyncio.gather(
+        #     *[self.kill_app(app) for app in victim_apps],
+        #     *[self.relinquish_entity(entity_id) for entity_id in victim_entity_ids]
+        # )

projects/sbos-playground/sbos/playground/schemas/__init__.py

@@ -40,4 +40,5 @@
     DomainResourceConstraintRead,
     ResourceConstraintRead,
     ResourceConstraintUpdate,
+    ResourceConstraintDelete,
 )