NAS-130513 / 24.10 / Add app events (#14173)

* Add basic service to get docker events

* Before pulling images validate docker state

* Send docker events for compose based projects

* Add util to get app name from project name

* Add basics for docker events

* Add implementation of app.events.process

* Setup docker events after service starts

* Top event is redundant

* Do not automatically send crud app events

* Make sure events are sent on app crud and upgrade/rollback

* Fix events being sent when converting from normal app to custom app

* Fix converting app job progress bits

* No need for redundant option

* Remove select as it was for dev purposes

* Event lock is not required

* Add docker_read role for docker events

src/middlewared/middlewared/plugins/apps/crud.py

@@ -25,6 +25,7 @@ class AppService(CRUDService):
     class Config:
         namespace = 'app'
         datastore_primary_key_type = 'string'
+        event_send = False
         cli_namespace = 'app'
         role_prefix = 'APPS'
 
@@ -221,6 +222,8 @@ def create_internal(
             new_values = add_context_to_values(app_name, new_values, app_version_details['app_metadata'], install=True)
             update_app_config(app_name, version, new_values)
             update_app_metadata(app_name, app_version_details, migrated_app)
+            # At this point the app exists
+            self.middleware.send_event('app.query', 'ADDED', id=app_name)
 
             job.set_progress(60, 'App installation in progress, pulling images')
             if dry_run is False:
@@ -234,6 +237,7 @@ def create_internal(
                 with contextlib.suppress(Exception):
                     method(*args, **kwargs)
 
+            self.middleware.send_event('app.query', 'REMOVED', id=app_name)
             raise e from None
         else:
             if dry_run is False:
@@ -290,6 +294,7 @@ def update_internal(self, job, app, data, progress_keyword='Update'):
             # TODO: Eventually we would want this to be executed for custom apps as well
             update_app_metadata_for_portals(app_name, app['version'])
         job.set_progress(60, 'Configuration updated, updating docker resources')
+        self.middleware.send_event('app.query', 'CHANGED', id=app_name)
         compose_action(app_name, app['version'], 'up', force_recreate=True, remove_orphans=True)
 
         job.set_progress(100, f'{progress_keyword} completed for {app_name!r}')
@@ -325,6 +330,9 @@ def delete_internal(self, job, app_name, app_config, options):
                 self.middleware.call_sync('zfs.dataset.delete', apps_volume_ds, {'recursive': True})
         finally:
             self.middleware.call_sync('app.metadata.generate').wait_sync(raise_error=True)
+
+        if options.get('send_event', True):
+            self.middleware.send_event('app.query', 'REMOVED', id=app_name)
         job.set_progress(100, f'Deleted {app_name!r} app')
         return True
 

src/middlewared/middlewared/plugins/apps/custom_app.py

@@ -37,21 +37,24 @@ def convert(self, job, app_name):
         job.set_progress(20, 'Removing existing app\'s docker resources')
         self.middleware.call_sync(
             'app.delete_internal', type('dummy_job', (object,), {'set_progress': lambda *args: None})(),
-            app_name, app, {'remove_images': False, 'remove_ix_volumes': False}
+            app_name, app, {'remove_images': False, 'remove_ix_volumes': False, 'send_event': False}
         )
 
         return self.create({
             'app_name': app_name,
             'custom_compose_config': rendered_config,
+            'conversion': True,
         }, job)
 
     def create(self, data, job=None, progress_base=0):
         """
         Create a custom app.
         """
         compose_config = validate_payload(data, 'app_create')
+        app_being_converted = data.get('conversion', False)
 
         def update_progress(percentage_done, message):
+            nonlocal progress_base
             job.set_progress(int((100 - progress_base) * (percentage_done / 100)) + progress_base, message)
 
         # For debug purposes
@@ -67,19 +70,32 @@ def update_progress(percentage_done, message):
             update_app_config(app_name, version, compose_config, custom_app=True)
             update_app_metadata(app_name, app_version_details, migrated=False, custom_app=True)
 
-            update_progress(60, 'App installation in progress, pulling images')
+            if app_being_converted is False:
+                self.middleware.send_event('app.query', 'ADDED', id=app_name)
+            if app_being_converted:
+                msg = 'App conversion in progress, pulling images'
+            else:
+                msg = 'App installation in progress, pulling images'
+            update_progress(60, msg)
             compose_action(app_name, version, 'up', force_recreate=True, remove_orphans=True)
         except Exception as e:
-            update_progress(80, f'Failure occurred while installing {app_name!r}, cleaning up')
+            update_progress(
+                80,
+                'Failure occurred while '
+                f'{"converting" if app_being_converted else "installing"} {app_name!r}, cleaning up'
+            )
             for method, args, kwargs in (
                 (compose_action, (app_name, version, 'down'), {'remove_orphans': True}),
                 (shutil.rmtree, (get_installed_app_path(app_name),), {}),
             ):
                 with contextlib.suppress(Exception):
                     method(*args, **kwargs)
 
+            self.middleware.send_event('app.query', 'REMOVED', id=app_name)
             raise e from None
         else:
             self.middleware.call_sync('app.metadata.generate').wait_sync(raise_error=True)
-            job.set_progress(100, f'{app_name!r} installed successfully')
+            job.set_progress(
+                100, f'{app_name!r} {"converted to custom app" if app_being_converted else "installed"} successfully'
+            )
             return self.middleware.call_sync('app.get_instance', app_name)

src/middlewared/middlewared/plugins/apps/events.py

@@ -0,0 +1,38 @@
+from middlewared.service import Service
+
+from .ix_apps.utils import get_app_name_from_project_name
+
+
+PROCESSING_APP_EVENT = set()
+
+
+class AppEvents(Service):
+
+    class Config:
+        namespace = 'app.events'
+        private = True
+
+    async def process(self, app_name, container_event):
+        if app := await self.middleware.call('app.query', [['id', '=', app_name]]):
+            self.middleware.send_event(
+                'app.query', 'CHANGED', id=app_name, fields=app[0],
+            )
+
+
+async def app_event(middleware, event_type, args):
+    app_name = get_app_name_from_project_name(args['id'])
+    if app_name in PROCESSING_APP_EVENT:
+        return
+
+    PROCESSING_APP_EVENT.add(app_name)
+
+    try:
+        await middleware.call('app.events.process', app_name, args['fields'])
+    except Exception as e:
+        middleware.logger.warning('Unhandled exception: %s', e)
+    finally:
+        PROCESSING_APP_EVENT.remove(app_name)
+
+
+async def setup(middleware):
+    middleware.event_subscribe('docker.events', app_event)

src/middlewared/middlewared/plugins/apps/ix_apps/query.py

@@ -6,7 +6,7 @@
 from .metadata import get_collective_config, get_collective_metadata
 from .lifecycle import get_current_app_config
 from .path import get_app_parent_config_path
-from .utils import PROJECT_PREFIX
+from .utils import PROJECT_PREFIX, get_app_name_from_project_name
 
 
 COMPOSE_SERVICE_KEY: str = 'com.docker.compose.service'
@@ -71,7 +71,7 @@ def list_apps(
     for app_name, app_resources in list_resources_by_project(
         project_name=f'{PROJECT_PREFIX}{specific_app}' if specific_app else None,
     ).items():
-        app_name = app_name[len(PROJECT_PREFIX):]
+        app_name = get_app_name_from_project_name(app_name)
         app_names.add(app_name)
         if app_name not in metadata:
             # The app is malformed or something is seriously wrong with it

src/middlewared/middlewared/plugins/apps/ix_apps/utils.py

@@ -1,3 +1,7 @@
 from catalog_reader.library import RE_VERSION  # noqa
 from middlewared.plugins.apps.schema_utils import CONTEXT_KEY_NAME  # noqa
 from middlewared.plugins.apps.utils import IX_APPS_MOUNT_PATH, PROJECT_PREFIX, run  # noqa
+
+
+def get_app_name_from_project_name(project_name: str) -> str:
+    return project_name[len(PROJECT_PREFIX):]

src/middlewared/middlewared/plugins/apps/rollback.py

@@ -61,6 +61,7 @@ def rollback(self, job, app_name, options):
         # 5) Roll back ix_volume dataset's snapshots if available
         # 6) Finally update collective metadata config to reflect new version
         update_app_metadata(app_name, rollback_version)
+        self.middleware.send_event('app.query', 'CHANGED', id=app_name)
         try:
             if options['rollback_snapshot'] and (
                 app_volume_ds := self.middleware.call_sync('app.get_app_volume_ds', app_name)

src/middlewared/middlewared/plugins/apps/upgrade.py

@@ -67,6 +67,7 @@ def upgrade(self, job, app_name, options):
 
             job.set_progress(40, f'Configuration updated for {app_name!r}, upgrading app')
 
+        self.middleware.send_event('app.query', 'CHANGED', id=app_name)
         try:
             compose_action(app_name, upgrade_version['version'], 'up', force_recreate=True, remove_orphans=True)
         finally:

src/middlewared/middlewared/plugins/apps_images/images.py

@@ -106,6 +106,7 @@ def callback(entry):
 
             job.set_progress((progress['current']/progress['total']) * 90, 'Pulling image')
 
+        self.middleware.call_sync('docker.state.validate')
         auth_config = data['auth_config'] or {}
         image_tag = data['image']
         pull_image(image_tag, callback, auth_config.get('username'), auth_config.get('password'))

src/middlewared/middlewared/plugins/docker/events.py

@@ -0,0 +1,47 @@
+from middlewared.plugins.apps.ix_apps.docker.utils import get_docker_client, PROJECT_KEY
+from middlewared.service import Service
+
+
+class DockerEventService(Service):
+
+    class Config:
+        namespace = 'docker.events'
+        private = True
+
+    def setup(self):
+        if not self.middleware.call_sync('docker.state.validate', False):
+            return
+
+        try:
+            self.process()
+        except Exception:
+            if not self.middleware.call('service.started', 'docker'):
+                # This is okay and can happen when docker is stopped
+                return
+            raise
+
+    def process(self):
+        with get_docker_client() as docker_client:
+            self.process_internal(docker_client)
+
+    def process_internal(self, client):
+        for container_event in client.events(
+            decode=True, filters={
+                'type': ['container'],
+                'event': [
+                    'create', 'destroy', 'detach', 'die', 'health_status', 'kill', 'unpause',
+                    'oom', 'pause', 'rename', 'resize', 'restart', 'start', 'stop', 'update',
+                ]
+            }
+        ):
+            if not isinstance(container_event, dict):
+                continue
+
+            if project := container_event.get('Actor', {}).get('Attributes', {}).get(PROJECT_KEY):
+                self.middleware.send_event('docker.events', 'ADDED', id=project, fields=container_event)
+
+
+async def setup(middleware):
+    middleware.event_register('docker.events', 'Docker container events', roles=['DOCKER_READ'])
+    # We are going to check in setup docker events if setting up events is relevant or not
+    middleware.create_task(middleware.call('docker.events.setup'))

src/middlewared/middlewared/plugins/service_/services/docker.py

@@ -60,6 +60,7 @@ async def stop(self):
 
     async def after_start(self):
         await self.middleware.call('docker.state.set_status', Status.RUNNING.value)
+        self.middleware.create_task(self.middleware.call('docker.events.setup'))
         await self.middleware.call('catalog.sync')
 
     async def before_stop(self):