add event for scaleset state updates (#882)

This moves all scaleset state updates through `Scaleset.set_state` and adds a new event EventScalesetStateUpdated.

src/api-service/__app__/onefuzzlib/autoscale.py

@@ -40,8 +40,7 @@ def scale_up(pool: Pool, scalesets: List[Scaleset], nodes_needed: int) -> None:
                 else:
                     scaleset.size = max_size
                     nodes_needed = nodes_needed - (max_size - current_size)
-                scaleset.state = ScalesetState.resize
-                scaleset.save()
+                scaleset.set_state(ScalesetState.resize)
 
             else:
                 continue
@@ -90,8 +89,7 @@ def scale_down(scalesets: List[Scaleset], nodes_to_remove: int) -> None:
             ScalesetState.shutdown,
             ScalesetState.halt,
         ]:
-            scaleset.state = ScalesetState.resize
-            scaleset.save()
+            scaleset.set_state(ScalesetState.resize)
 
         free_nodes = Node.search_states(
             scaleset_id=scaleset.scaleset_id,
@@ -107,9 +105,8 @@ def scale_down(scalesets: List[Scaleset], nodes_to_remove: int) -> None:
             max_nodes_remove = min(len(nodes), nodes_to_remove)
             # All nodes in scaleset are free. Can shutdown VMSS
             if max_nodes_remove >= scaleset.size and len(nodes) >= scaleset.size:
-                scaleset.state = ScalesetState.shutdown
+                scaleset.set_state(ScalesetState.shutdown)
                 nodes_to_remove = nodes_to_remove - scaleset.size
-                scaleset.save()
                 for node in nodes:
                     node.set_shutdown()
                 continue
@@ -117,8 +114,7 @@ def scale_down(scalesets: List[Scaleset], nodes_to_remove: int) -> None:
             # Resize of VMSS needed
             scaleset.size = scaleset.size - max_nodes_remove
             nodes_to_remove = nodes_to_remove - max_nodes_remove
-            scaleset.state = ScalesetState.resize
-            scaleset.save()
+            scaleset.set_state(ScalesetState.resize)
 
 
 def get_vm_count(tasks: List[Task]) -> int:

src/api-service/__app__/onefuzzlib/workers/pools.py

@@ -222,8 +222,7 @@ class Pool(BASE_POOL, ORMMixin):
             return
 
         for scaleset in scalesets:
-            scaleset.state = ScalesetState.halt
-            scaleset.save()
+            scaleset.set_state(ScalesetState.halt)
 
         for node in nodes:
             node.set_halt()

src/api-service/__app__/onefuzzlib/workers/scalesets.py

@@ -13,6 +13,7 @@ from onefuzztypes.events import (
     EventScalesetCreated,
     EventScalesetDeleted,
     EventScalesetFailed,
+    EventScalesetStateUpdated,
 )
 from onefuzztypes.models import Error
 from onefuzztypes.models import Scaleset as BASE_SCALESET
@@ -141,8 +142,7 @@ class Scaleset(BASE_SCALESET, ORMMixin):
             return
 
         self.error = error
-        self.state = ScalesetState.creation_failed
-        self.save()
+        self.set_state(ScalesetState.creation_failed)
 
         send_event(
             EventScalesetFailed(
@@ -184,11 +184,9 @@ class Scaleset(BASE_SCALESET, ORMMixin):
                 self.set_failed(error)
                 return
             else:
-                self.state = ScalesetState.setup
+                self.set_state(ScalesetState.setup)
         else:
-            self.state = ScalesetState.setup
-
-        self.save()
+            self.set_state(ScalesetState.setup)
 
     def setup(self) -> None:
         from .pools import Pool
@@ -269,7 +267,7 @@ class Scaleset(BASE_SCALESET, ORMMixin):
                 self.set_failed(identity_result)
                 return
             else:
-                self.state = ScalesetState.running
+                self.set_state(ScalesetState.running)
         self.save()
 
     def try_set_identity(self, vmss: Any) -> Optional[Error]:
@@ -414,9 +412,7 @@ class Scaleset(BASE_SCALESET, ORMMixin):
         node_count = len(Node.search_states(scaleset_id=self.scaleset_id))
         if node_count == self.size:
             logging.info(SCALESET_LOG_PREFIX + "resize finished: %s", self.scaleset_id)
-            self.state = ScalesetState.running
-            self.save()
-            return
+            self.set_state(ScalesetState.running)
         else:
             logging.info(
                 SCALESET_LOG_PREFIX
@@ -426,7 +422,6 @@ class Scaleset(BASE_SCALESET, ORMMixin):
                 node_count,
                 self.size,
             )
-            return
 
     def _resize_grow(self) -> None:
         try:
@@ -577,10 +572,16 @@ class Scaleset(BASE_SCALESET, ORMMixin):
         if self.state in [ScalesetState.halt, ScalesetState.shutdown]:
             return
 
+        logging.info(
+            SCALESET_LOG_PREFIX + "scaleset set_shutdown: scaleset_id:%s now:%s",
+            self.scaleset_id,
+            now,
+        )
+
         if now:
-            self.state = ScalesetState.halt
+            self.set_state(ScalesetState.halt)
         else:
-            self.state = ScalesetState.shutdown
+            self.set_state(ScalesetState.shutdown)
 
         self.save()
 
@@ -736,6 +737,18 @@ class Scaleset(BASE_SCALESET, ORMMixin):
             EventScalesetDeleted(scaleset_id=self.scaleset_id, pool_name=self.pool_name)
         )
 
+    def set_state(self, state: ScalesetState) -> None:
+        if self.state == state:
+            return
+
+        self.state = state
+        self.save()
+        send_event(
+            EventScalesetStateUpdated(
+                scaleset_id=self.scaleset_id, pool_name=self.pool_name, state=self.state
+            )
+        )
+
 
 class ShrinkEntry(BaseModel):
     shrink_id: UUID = Field(default_factory=uuid4)