Observer plugin (#2037)

* Observer plugin

* fix error handling

* remove stop function

* fix restart policy

* Add observer watchdog

* Add observer

* Update supervisor/plugins/observer.py

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

* Expose port 4357

Co-authored-by: Joakim Sørensen <joasoe@gmail.com>

API.md

@@ -946,6 +946,41 @@ return:
 }
 ```
 
+### Observer
+
+- GET `/observer/info`
+
+```json
+{
+  "host": "ip-address",
+  "version": "1",
+  "version_latest": "2"
+}
+```
+
+- POST `/observer/update`
+
+```json
+{
+  "version": "VERSION"
+}
+```
+
+- GET `/observer/stats`
+
+```json
+{
+  "cpu_percent": 0.0,
+  "memory_usage": 283123,
+  "memory_limit": 329392,
+  "memory_percent": 1.4,
+  "network_tx": 0,
+  "network_rx": 0,
+  "blk_read": 0,
+  "blk_write": 0
+}
+```
+
 ### Multicast
 
 - GET `/multicast/info`

supervisor/api/__init__.py

@@ -19,6 +19,7 @@ from .info import APIInfo
 from .ingress import APIIngress
 from .multicast import APIMulticast
 from .network import APINetwork
+from .observer import APIObserver
 from .os import APIOS
 from .proxy import APIProxy
 from .security import SecurityMiddleware
@@ -54,6 +55,7 @@ class RestAPI(CoreSysAttributes):
         self._register_host()
         self._register_os()
         self._register_cli()
+        self._register_observer()
         self._register_multicast()
         self._register_network()
         self._register_hardware()
@@ -135,6 +137,19 @@ class RestAPI(CoreSysAttributes):
             ]
         )
 
+    def _register_observer(self) -> None:
+        """Register Observer functions."""
+        api_observer = APIObserver()
+        api_observer.coresys = self.coresys
+
+        self.webapp.add_routes(
+            [
+                web.get("/observer/info", api_observer.info),
+                web.get("/observer/stats", api_observer.stats),
+                web.post("/observer/update", api_observer.update),
+            ]
+        )
+
     def _register_multicast(self) -> None:
         """Register Multicast functions."""
         api_multicast = APIMulticast()

supervisor/api/observer.py

@@ -0,0 +1,65 @@
+"""Init file for Supervisor Observer RESTful API."""
+import asyncio
+import logging
+from typing import Any, Dict
+
+from aiohttp import web
+import voluptuous as vol
+
+from ..const import (
+    ATTR_BLK_READ,
+    ATTR_BLK_WRITE,
+    ATTR_CPU_PERCENT,
+    ATTR_HOST,
+    ATTR_MEMORY_LIMIT,
+    ATTR_MEMORY_PERCENT,
+    ATTR_MEMORY_USAGE,
+    ATTR_NETWORK_RX,
+    ATTR_NETWORK_TX,
+    ATTR_VERSION,
+    ATTR_VERSION_LATEST,
+)
+from ..coresys import CoreSysAttributes
+from ..validate import version_tag
+from .utils import api_process, api_validate
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
+
+
+class APIObserver(CoreSysAttributes):
+    """Handle RESTful API for Observer functions."""
+
+    @api_process
+    async def info(self, request: web.Request) -> Dict[str, Any]:
+        """Return HA Observer information."""
+        return {
+            ATTR_HOST: str(self.sys_docker.network.observer),
+            ATTR_VERSION: self.sys_plugins.observer.version,
+            ATTR_VERSION_LATEST: self.sys_plugins.observer.latest_version,
+        }
+
+    @api_process
+    async def stats(self, request: web.Request) -> Dict[str, Any]:
+        """Return resource information."""
+        stats = await self.sys_plugins.observer.stats()
+
+        return {
+            ATTR_CPU_PERCENT: stats.cpu_percent,
+            ATTR_MEMORY_USAGE: stats.memory_usage,
+            ATTR_MEMORY_LIMIT: stats.memory_limit,
+            ATTR_MEMORY_PERCENT: stats.memory_percent,
+            ATTR_NETWORK_RX: stats.network_rx,
+            ATTR_NETWORK_TX: stats.network_tx,
+            ATTR_BLK_READ: stats.blk_read,
+            ATTR_BLK_WRITE: stats.blk_write,
+        }
+
+    @api_process
+    async def update(self, request: web.Request) -> None:
+        """Update HA observer."""
+        body = await api_validate(SCHEMA_VERSION, request)
+        version = body.get(ATTR_VERSION, self.sys_plugins.observer.latest_version)
+
+        await asyncio.shield(self.sys_plugins.observer.update(version))

supervisor/const.py

@@ -17,15 +17,11 @@ URL_HASSOS_OTA = (
 SUPERVISOR_DATA = Path("/data")
 
 FILE_HASSIO_ADDONS = Path(SUPERVISOR_DATA, "addons.json")
-FILE_HASSIO_AUDIO = Path(SUPERVISOR_DATA, "audio.json")
 FILE_HASSIO_AUTH = Path(SUPERVISOR_DATA, "auth.json")
-FILE_HASSIO_CLI = Path(SUPERVISOR_DATA, "cli.json")
 FILE_HASSIO_CONFIG = Path(SUPERVISOR_DATA, "config.json")
 FILE_HASSIO_DISCOVERY = Path(SUPERVISOR_DATA, "discovery.json")
-FILE_HASSIO_DNS = Path(SUPERVISOR_DATA, "dns.json")
 FILE_HASSIO_HOMEASSISTANT = Path(SUPERVISOR_DATA, "homeassistant.json")
 FILE_HASSIO_INGRESS = Path(SUPERVISOR_DATA, "ingress.json")
-FILE_HASSIO_MULTICAST = Path(SUPERVISOR_DATA, "multicast.json")
 FILE_HASSIO_SERVICES = Path(SUPERVISOR_DATA, "services.json")
 FILE_HASSIO_UPDATER = Path(SUPERVISOR_DATA, "updater.json")
 
@@ -75,6 +71,7 @@ HEADER_TOKEN_OLD = "X-Hassio-Key"
 ENV_TIME = "TZ"
 ENV_TOKEN = "SUPERVISOR_TOKEN"
 ENV_TOKEN_OLD = "HASSIO_TOKEN"
+ENV_OBSERVER = "OBSERVER_TOKEN"
 
 ENV_HOMEASSISTANT_REPOSITORY = "HOMEASSISTANT_REPOSITORY"
 ENV_SUPERVISOR_DEV = "SUPERVISOR_DEV"
@@ -275,6 +272,7 @@ ATTR_VPN = "vpn"
 ATTR_WAIT_BOOT = "wait_boot"
 ATTR_WATCHDOG = "watchdog"
 ATTR_WEBUI = "webui"
+ATTR_OBSERVER = "observer"
 
 PROVIDE_SERVICE = "provide"
 NEED_SERVICE = "need"

supervisor/docker/cli.py

@@ -1,7 +1,7 @@
 """HA Cli docker object."""
 import logging
 
-from ..const import ENV_TIME, ENV_TOKEN
+from ..const import ENV_OBSERVER, ENV_TIME, ENV_TOKEN
 from ..coresys import CoreSysAttributes
 from .interface import DockerInterface
 
@@ -45,10 +45,14 @@ class DockerCli(DockerInterface, CoreSysAttributes):
             name=self.name,
             hostname=self.name.replace("_", "-"),
             detach=True,
-            extra_hosts={"supervisor": self.sys_docker.network.supervisor},
+            extra_hosts={
+                "supervisor": self.sys_docker.network.supervisor,
+                "observer": self.sys_docker.network.observer,
+            },
             environment={
                 ENV_TIME: self.sys_config.timezone,
                 ENV_TOKEN: self.sys_plugins.cli.supervisor_token,
+                ENV_OBSERVER: self.sys_plugins.observer.access_token,
             },
         )
 

supervisor/docker/homeassistant.py

@@ -6,7 +6,14 @@ from typing import Awaitable, Dict, Optional
 import docker
 import requests
 
-from ..const import ENV_TIME, ENV_TOKEN, ENV_TOKEN_OLD, LABEL_MACHINE, MACHINE_ID
+from ..const import (
+    ENV_OBSERVER,
+    ENV_TIME,
+    ENV_TOKEN,
+    ENV_TOKEN_OLD,
+    LABEL_MACHINE,
+    MACHINE_ID,
+)
 from ..exceptions import DockerAPIError
 from .interface import CommandReturn, DockerInterface
 
@@ -115,12 +122,17 @@ class DockerHomeAssistant(DockerInterface):
             init=False,
             network_mode="host",
             volumes=self.volumes,
+            extra_hosts={
+                "supervisor": self.sys_docker.network.supervisor,
+                "observer": self.sys_docker.network.observer,
+            },
             environment={
                 "HASSIO": self.sys_docker.network.supervisor,
                 "SUPERVISOR": self.sys_docker.network.supervisor,
                 ENV_TIME: self.sys_config.timezone,
                 ENV_TOKEN: self.sys_homeassistant.supervisor_token,
                 ENV_TOKEN_OLD: self.sys_homeassistant.supervisor_token,
+                ENV_OBSERVER: self.sys_plugins.observer.access_token,
             },
         )
 

supervisor/docker/network.py

@@ -69,6 +69,11 @@ class DockerNetwork:
         """Return cli of the network."""
         return DOCKER_NETWORK_MASK[5]
 
+    @property
+    def observer(self) -> IPv4Address:
+        """Return observer of the network."""
+        return DOCKER_NETWORK_MASK[6]
+
     def _get_network(self) -> docker.models.networks.Network:
         """Get supervisor network."""
         try:

supervisor/docker/observer.py

@@ -0,0 +1,62 @@
+"""Observer docker object."""
+import logging
+
+from ..const import ENV_OBSERVER, ENV_TIME
+from ..coresys import CoreSysAttributes
+from .interface import DockerInterface
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+OBSERVER_DOCKER_NAME: str = "hassio_observer"
+
+
+class DockerObserver(DockerInterface, CoreSysAttributes):
+    """Docker Supervisor wrapper for observer plugin."""
+
+    @property
+    def image(self):
+        """Return name of observer image."""
+        return self.sys_plugins.observer.image
+
+    @property
+    def name(self) -> str:
+        """Return name of Docker container."""
+        return OBSERVER_DOCKER_NAME
+
+    def _run(self) -> None:
+        """Run Docker image.
+
+        Need run inside executor.
+        """
+        if self._is_running():
+            return
+
+        # Cleanup
+        self._stop()
+
+        # Create & Run container
+        docker_container = self.sys_docker.run(
+            self.image,
+            version=self.sys_plugins.observer.version,
+            init=False,
+            ipv4=self.sys_docker.network.observer,
+            name=self.name,
+            hostname=self.name.replace("_", "-"),
+            detach=True,
+            restart_policy={"Name": "always"},
+            extra_hosts={"supervisor": self.sys_docker.network.supervisor},
+            environment={
+                ENV_TIME: self.sys_config.timezone,
+                ENV_OBSERVER: self.sys_plugins.observer.access_token,
+            },
+            volumes={"/run/docker.sock": {"bind": "/run/docker.sock", "mode": "ro"}},
+            ports={"80/tcp": 4357},
+        )
+
+        self._meta = docker_container.attrs
+        _LOGGER.info(
+            "Start Observer %s with version %s - %s",
+            self.image,
+            self.version,
+            self.sys_docker.network.observer,
+        )

supervisor/exceptions.py

@@ -65,6 +65,17 @@ class CliUpdateError(CliError):
     """Error on update of a HA cli."""
 
 
+# Observer
+
+
+class ObserverError(HassioError):
+    """General Observer exception."""
+
+
+class ObserverUpdateError(ObserverError):
+    """Error on update of a Observer."""
+
+
 # Multicast
 
 

supervisor/misc/tasks.py

@@ -10,6 +10,7 @@ from ..exceptions import (
     CoreDNSError,
     HomeAssistantError,
     MulticastError,
+    ObserverError,
 )
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -22,6 +23,7 @@ RUN_UPDATE_CLI = 28100
 RUN_UPDATE_DNS = 30100
 RUN_UPDATE_AUDIO = 30200
 RUN_UPDATE_MULTICAST = 30300
+RUN_UPDATE_OBSERVER = 30400
 
 RUN_RELOAD_ADDONS = 10800
 RUN_RELOAD_SNAPSHOTS = 72000
@@ -35,6 +37,7 @@ RUN_WATCHDOG_HOMEASSISTANT_API = 120
 RUN_WATCHDOG_DNS_DOCKER = 30
 RUN_WATCHDOG_AUDIO_DOCKER = 60
 RUN_WATCHDOG_CLI_DOCKER = 60
+RUN_WATCHDOG_OBSERVER_DOCKER = 60
 RUN_WATCHDOG_MULTICAST_DOCKER = 60
 
 RUN_WATCHDOG_ADDON_DOCKER = 30
@@ -60,6 +63,7 @@ class Tasks(CoreSysAttributes):
         self.sys_scheduler.register_task(self._update_dns, RUN_UPDATE_DNS)
         self.sys_scheduler.register_task(self._update_audio, RUN_UPDATE_AUDIO)
         self.sys_scheduler.register_task(self._update_multicast, RUN_UPDATE_MULTICAST)
+        self.sys_scheduler.register_task(self._update_observer, RUN_UPDATE_OBSERVER)
 
         # Reload
         self.sys_scheduler.register_task(self.sys_store.reload, RUN_RELOAD_ADDONS)
@@ -86,6 +90,9 @@ class Tasks(CoreSysAttributes):
         self.sys_scheduler.register_task(
             self._watchdog_cli_docker, RUN_WATCHDOG_CLI_DOCKER
         )
+        self.sys_scheduler.register_task(
+            self._watchdog_observer_docker, RUN_WATCHDOG_OBSERVER_DOCKER
+        )
         self.sys_scheduler.register_task(
             self._watchdog_multicast_docker, RUN_WATCHDOG_MULTICAST_DOCKER
         )
@@ -225,6 +232,14 @@ class Tasks(CoreSysAttributes):
         _LOGGER.info("Found new PulseAudio plugin version")
         await self.sys_plugins.audio.update()
 
+    async def _update_observer(self):
+        """Check and run update of Observer plugin."""
+        if not self.sys_plugins.observer.need_update:
+            return
+
+        _LOGGER.info("Found new Observer plugin version")
+        await self.sys_plugins.observer.update()
+
     async def _update_multicast(self):
         """Check and run update of multicast."""
         if not self.sys_plugins.multicast.need_update:
@@ -278,6 +293,21 @@ class Tasks(CoreSysAttributes):
         except CliError:
             _LOGGER.error("Watchdog cli reanimation failed!")
 
+    async def _watchdog_observer_docker(self):
+        """Check running state of Docker and start if they is close."""
+        # if observer plugin is active
+        if (
+            await self.sys_plugins.observer.is_running()
+            or self.sys_plugins.observer.in_progress
+        ):
+            return
+        _LOGGER.warning("Watchdog found a problem with observer plugin!")
+
+        try:
+            await self.sys_plugins.observer.start()
+        except ObserverError:
+            _LOGGER.error("Watchdog observer reanimation failed!")
+
     async def _watchdog_multicast_docker(self):
         """Check running state of Docker and start if they is close."""
         # if multicast plugin is active

supervisor/plugins/__init__.py

@@ -10,6 +10,7 @@ from .audio import Audio
 from .cli import HaCli
 from .dns import CoreDNS
 from .multicast import Multicast
+from .observer import Observer
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -20,6 +21,7 @@ class PluginManager(CoreSysAttributes):
     required_cli: LegacyVersion = pkg_parse("26")
     required_dns: LegacyVersion = pkg_parse("9")
     required_audio: LegacyVersion = pkg_parse("17")
+    required_observer: LegacyVersion = pkg_parse("1")
     required_multicast: LegacyVersion = pkg_parse("3")
 
     def __init__(self, coresys: CoreSys):
@@ -29,6 +31,7 @@ class PluginManager(CoreSysAttributes):
         self._cli: HaCli = HaCli(coresys)
         self._dns: CoreDNS = CoreDNS(coresys)
         self._audio: Audio = Audio(coresys)
+        self._observer: Observer = Observer(coresys)
         self._multicast: Multicast = Multicast(coresys)
 
     @property
@@ -46,6 +49,11 @@ class PluginManager(CoreSysAttributes):
         """Return audio handler."""
         return self._audio
 
+    @property
+    def observer(self) -> Observer:
+        """Return observer handler."""
+        return self._observer
+
     @property
     def multicast(self) -> Multicast:
         """Return multicast handler."""
@@ -58,6 +66,7 @@ class PluginManager(CoreSysAttributes):
             self.dns,
             self.audio,
             self.cli,
+            self.observer,
             self.multicast,
         ):
             try:
@@ -71,6 +80,7 @@ class PluginManager(CoreSysAttributes):
             (self._audio, self.required_audio),
             (self._dns, self.required_dns),
             (self._cli, self.required_cli),
+            (self._observer, self.required_observer),
             (self._multicast, self.required_multicast),
         ):
             # Check if need an update
@@ -109,6 +119,7 @@ class PluginManager(CoreSysAttributes):
                 self.dns.repair(),
                 self.audio.repair(),
                 self.cli.repair(),
+                self.observer.repair(),
                 self.multicast.repair(),
             ]
         )

supervisor/plugins/audio.py

@@ -11,12 +11,13 @@ from typing import Awaitable, Optional
 
 import jinja2
 
-from ..const import ATTR_IMAGE, ATTR_VERSION, FILE_HASSIO_AUDIO
+from ..const import ATTR_IMAGE, ATTR_VERSION
 from ..coresys import CoreSys, CoreSysAttributes
 from ..docker.audio import DockerAudio
 from ..docker.stats import DockerStats
 from ..exceptions import AudioError, AudioUpdateError, DockerAPIError
 from ..utils.json import JsonConfig
+from .const import FILE_HASSIO_AUDIO
 from .validate import SCHEMA_AUDIO_CONFIG
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -225,8 +226,9 @@ class Audio(JsonConfig, CoreSysAttributes):
         _LOGGER.info("Repair Audio %s", self.version)
         try:
             await self.instance.install(self.version)
-        except DockerAPIError:
+        except DockerAPIError as err:
             _LOGGER.error("Repairing of Audio failed")
+            self.sys_capture_exception(err)
 
     def pulse_client(self, input_profile=None, output_profile=None) -> str:
         """Generate an /etc/pulse/client.conf data."""

supervisor/plugins/cli.py

@@ -8,12 +8,13 @@ import logging
 import secrets
 from typing import Awaitable, Optional
 
-from ..const import ATTR_ACCESS_TOKEN, ATTR_IMAGE, ATTR_VERSION, FILE_HASSIO_CLI
+from ..const import ATTR_ACCESS_TOKEN, ATTR_IMAGE, ATTR_VERSION
 from ..coresys import CoreSys, CoreSysAttributes
 from ..docker.cli import DockerCli
 from ..docker.stats import DockerStats
 from ..exceptions import CliError, CliUpdateError, DockerAPIError
 from ..utils.json import JsonConfig
+from .const import FILE_HASSIO_CLI
 from .validate import SCHEMA_CLI_CONFIG
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -90,7 +91,7 @@ class HaCli(CoreSysAttributes, JsonConfig):
             self.image = self.instance.image
             self.save_data()
 
-        # Run PulseAudio
+        # Run CLI
         with suppress(CliError):
             if not await self.instance.is_running():
                 await self.start()
@@ -192,5 +193,6 @@ class HaCli(CoreSysAttributes, JsonConfig):
         _LOGGER.info("Repair HA cli %s", self.version)
         try:
             await self.instance.install(self.version, latest=True)
-        except DockerAPIError:
+        except DockerAPIError as err:
             _LOGGER.error("Repairing of HA cli failed")
+            self.sys_capture_exception(err)

supervisor/plugins/const.py

@@ -0,0 +1,10 @@
+"""Const for plugins."""
+from pathlib import Path
+
+from ..const import SUPERVISOR_DATA
+
+FILE_HASSIO_AUDIO = Path(SUPERVISOR_DATA, "audio.json")
+FILE_HASSIO_CLI = Path(SUPERVISOR_DATA, "cli.json")
+FILE_HASSIO_DNS = Path(SUPERVISOR_DATA, "dns.json")
+FILE_HASSIO_OBSERVER = Path(SUPERVISOR_DATA, "observer.json")
+FILE_HASSIO_MULTICAST = Path(SUPERVISOR_DATA, "multicast.json")

supervisor/plugins/dns.py

@@ -13,20 +13,14 @@ import attr
 import jinja2
 import voluptuous as vol
 
-from ..const import (
-    ATTR_IMAGE,
-    ATTR_SERVERS,
-    ATTR_VERSION,
-    DNS_SUFFIX,
-    FILE_HASSIO_DNS,
-    LogLevel,
-)
+from ..const import ATTR_IMAGE, ATTR_SERVERS, ATTR_VERSION, DNS_SUFFIX, LogLevel
 from ..coresys import CoreSys, CoreSysAttributes
 from ..docker.dns import DockerDNS
 from ..docker.stats import DockerStats
 from ..exceptions import CoreDNSError, CoreDNSUpdateError, DockerAPIError
 from ..utils.json import JsonConfig
 from ..validate import dns_url
+from .const import FILE_HASSIO_DNS
 from .validate import SCHEMA_DNS_CONFIG
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -322,6 +316,7 @@ class CoreDNS(JsonConfig, CoreSysAttributes):
             write=False,
         )
         self.add_host(self.sys_docker.network.dns, ["dns"], write=False)
+        self.add_host(self.sys_docker.network.observer, ["observer"], write=False)
 
     def write_hosts(self) -> None:
         """Write hosts from memory to file."""
@@ -419,8 +414,9 @@ class CoreDNS(JsonConfig, CoreSysAttributes):
         _LOGGER.info("Repair CoreDNS %s", self.version)
         try:
             await self.instance.install(self.version)
-        except DockerAPIError:
+        except DockerAPIError as err:
             _LOGGER.error("Repairing of CoreDNS failed")
+            self.sys_capture_exception(err)
 
     def _write_resolv(self, resolv_conf: Path) -> None:
         """Update/Write resolv.conf file."""

supervisor/plugins/multicast.py

@@ -7,12 +7,13 @@ from contextlib import suppress
 import logging
 from typing import Awaitable, Optional
 
-from ..const import ATTR_IMAGE, ATTR_VERSION, FILE_HASSIO_MULTICAST
+from ..const import ATTR_IMAGE, ATTR_VERSION
 from ..coresys import CoreSys, CoreSysAttributes
 from ..docker.multicast import DockerMulticast
 from ..docker.stats import DockerStats
 from ..exceptions import DockerAPIError, MulticastError, MulticastUpdateError
 from ..utils.json import JsonConfig
+from .const import FILE_HASSIO_MULTICAST
 from .validate import SCHEMA_MULTICAST_CONFIG
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -205,5 +206,6 @@ class Multicast(JsonConfig, CoreSysAttributes):
         _LOGGER.info("Repair Multicast %s", self.version)
         try:
             await self.instance.install(self.version)
-        except DockerAPIError:
+        except DockerAPIError as err:
             _LOGGER.error("Repairing of Multicast failed")
+            self.sys_capture_exception(err)

supervisor/plugins/observer.py

@@ -0,0 +1,188 @@
+"""Home Assistant observer plugin.
+
+Code: https://github.com/home-assistant/plugin-observer
+"""
+import asyncio
+from contextlib import suppress
+import logging
+import secrets
+from typing import Awaitable, Optional
+
+from ..const import ATTR_ACCESS_TOKEN, ATTR_IMAGE, ATTR_VERSION
+from ..coresys import CoreSys, CoreSysAttributes
+from ..docker.observer import DockerObserver
+from ..docker.stats import DockerStats
+from ..exceptions import DockerAPIError, ObserverError, ObserverUpdateError
+from ..utils.json import JsonConfig
+from .const import FILE_HASSIO_OBSERVER
+from .validate import SCHEMA_OBSERVER_CONFIG
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+
+class Observer(CoreSysAttributes, JsonConfig):
+    """Supervisor observer instance."""
+
+    def __init__(self, coresys: CoreSys):
+        """Initialize observer handler."""
+        super().__init__(FILE_HASSIO_OBSERVER, SCHEMA_OBSERVER_CONFIG)
+        self.coresys: CoreSys = coresys
+        self.instance: DockerObserver = DockerObserver(coresys)
+
+    @property
+    def version(self) -> Optional[str]:
+        """Return version of observer."""
+        return self._data.get(ATTR_VERSION)
+
+    @version.setter
+    def version(self, value: str) -> None:
+        """Set current version of observer."""
+        self._data[ATTR_VERSION] = value
+
+    @property
+    def image(self) -> str:
+        """Return current image of observer."""
+        if self._data.get(ATTR_IMAGE):
+            return self._data[ATTR_IMAGE]
+        return f"homeassistant/{self.sys_arch.supervisor}-hassio-observer"
+
+    @image.setter
+    def image(self, value: str) -> None:
+        """Return current image of observer."""
+        self._data[ATTR_IMAGE] = value
+
+    @property
+    def latest_version(self) -> str:
+        """Return version of latest observer."""
+        return self.sys_updater.version_observer
+
+    @property
+    def need_update(self) -> bool:
+        """Return true if a observer update is available."""
+        return self.version != self.latest_version
+
+    @property
+    def access_token(self) -> str:
+        """Return an access token for the Observer API."""
+        return self._data.get(ATTR_ACCESS_TOKEN)
+
+    @property
+    def in_progress(self) -> bool:
+        """Return True if a task is in progress."""
+        return self.instance.in_progress
+
+    async def load(self) -> None:
+        """Load observer setup."""
+        # Check observer state
+        try:
+            # Evaluate Version if we lost this information
+            if not self.version:
+                self.version = await self.instance.get_latest_version()
+
+            await self.instance.attach(tag=self.version)
+        except DockerAPIError:
+            _LOGGER.info(
+                "No observer plugin Docker image %s found.", self.instance.image
+            )
+
+            # Install observer
+            with suppress(ObserverError):
+                await self.install()
+        else:
+            self.version = self.instance.version
+            self.image = self.instance.image
+            self.save_data()
+
+        # Run Observer
+        with suppress(ObserverError):
+            if not await self.instance.is_running():
+                await self.start()
+
+    async def install(self) -> None:
+        """Install observer."""
+        _LOGGER.info("Setup observer plugin")
+        while True:
+            # read observer tag and install it
+            if not self.latest_version:
+                await self.sys_updater.reload()
+
+            if self.latest_version:
+                with suppress(DockerAPIError):
+                    await self.instance.install(
+                        self.latest_version, image=self.sys_updater.image_observer
+                    )
+                    break
+            _LOGGER.warning("Error on install observer plugin. Retry in 30sec")
+            await asyncio.sleep(30)
+
+        _LOGGER.info("observer plugin now installed")
+        self.version = self.instance.version
+        self.image = self.sys_updater.image_observer
+        self.save_data()
+
+    async def update(self, version: Optional[str] = None) -> None:
+        """Update local HA observer."""
+        version = version or self.latest_version
+        old_image = self.image
+
+        if version == self.version:
+            _LOGGER.warning("Version %s is already installed for observer", version)
+            return
+
+        try:
+            await self.instance.update(version, image=self.sys_updater.image_observer)
+        except DockerAPIError as err:
+            _LOGGER.error("HA observer update failed")
+            raise ObserverUpdateError() from err
+        else:
+            self.version = version
+            self.image = self.sys_updater.image_observer
+            self.save_data()
+
+        # Cleanup
+        with suppress(DockerAPIError):
+            await self.instance.cleanup(old_image=old_image)
+
+        # Start observer
+        await self.start()
+
+    async def start(self) -> None:
+        """Run observer."""
+        # Create new API token
+        if not self.access_token:
+            self._data[ATTR_ACCESS_TOKEN] = secrets.token_hex(56)
+            self.save_data()
+
+        # Start Instance
+        _LOGGER.info("Start observer plugin")
+        try:
+            await self.instance.run()
+        except DockerAPIError as err:
+            _LOGGER.error("Can't start observer plugin")
+            raise ObserverError() from err
+
+    async def stats(self) -> DockerStats:
+        """Return stats of observer."""
+        try:
+            return await self.instance.stats()
+        except DockerAPIError as err:
+            raise ObserverError() from err
+
+    def is_running(self) -> Awaitable[bool]:
+        """Return True if Docker container is running.
+
+        Return a coroutine.
+        """
+        return self.instance.is_running()
+
+    async def repair(self) -> None:
+        """Repair observer container."""
+        if await self.instance.exists():
+            return
+
+        _LOGGER.info("Repair HA observer %s", self.version)
+        try:
+            await self.instance.install(self.version)
+        except DockerAPIError as err:
+            _LOGGER.error("Repairing of HA observer failed")
+            self.sys_capture_exception(err)

supervisor/plugins/validate.py

@@ -35,3 +35,13 @@ SCHEMA_MULTICAST_CONFIG = vol.Schema(
     {vol.Optional(ATTR_VERSION): version_tag, vol.Optional(ATTR_IMAGE): docker_image},
     extra=vol.REMOVE_EXTRA,
 )
+
+
+SCHEMA_OBSERVER_CONFIG = vol.Schema(
+    {
+        vol.Optional(ATTR_VERSION): version_tag,
+        vol.Optional(ATTR_IMAGE): docker_image,
+        vol.Optional(ATTR_ACCESS_TOKEN): token,
+    },
+    extra=vol.REMOVE_EXTRA,
+)

supervisor/updater.py

@@ -17,6 +17,7 @@ from .const import (
     ATTR_HOMEASSISTANT,
     ATTR_IMAGE,
     ATTR_MULTICAST,
+    ATTR_OBSERVER,
     ATTR_SUPERVISOR,
     FILE_HASSIO_UPDATER,
     URL_HASSIO_VERSION,
@@ -79,6 +80,11 @@ class Updater(JsonConfig, CoreSysAttributes):
         """Return latest version of Audio."""
         return self._data.get(ATTR_AUDIO)
 
+    @property
+    def version_observer(self) -> Optional[str]:
+        """Return latest version of Observer."""
+        return self._data.get(ATTR_OBSERVER)
+
     @property
     def version_multicast(self) -> Optional[str]:
         """Return latest version of Multicast."""
@@ -123,6 +129,15 @@ class Updater(JsonConfig, CoreSysAttributes):
             return None
         return self._data[ATTR_IMAGE][ATTR_AUDIO].format(arch=self.sys_arch.supervisor)
 
+    @property
+    def image_observer(self) -> Optional[str]:
+        """Return latest version of Observer."""
+        if ATTR_OBSERVER not in self._data[ATTR_IMAGE]:
+            return None
+        return self._data[ATTR_IMAGE][ATTR_OBSERVER].format(
+            arch=self.sys_arch.supervisor
+        )
+
     @property
     def image_multicast(self) -> Optional[str]:
         """Return latest version of Multicast."""
@@ -184,6 +199,7 @@ class Updater(JsonConfig, CoreSysAttributes):
             self._data[ATTR_CLI] = data["cli"]
             self._data[ATTR_DNS] = data["dns"]
             self._data[ATTR_AUDIO] = data["audio"]
+            self._data[ATTR_OBSERVER] = data["observer"]
             self._data[ATTR_MULTICAST] = data["multicast"]
 
             # Update images for that versions
@@ -192,6 +208,7 @@ class Updater(JsonConfig, CoreSysAttributes):
             self._data[ATTR_IMAGE][ATTR_AUDIO] = data["image"]["audio"]
             self._data[ATTR_IMAGE][ATTR_CLI] = data["image"]["cli"]
             self._data[ATTR_IMAGE][ATTR_DNS] = data["image"]["dns"]
+            self._data[ATTR_IMAGE][ATTR_OBSERVER] = data["image"]["observer"]
             self._data[ATTR_IMAGE][ATTR_MULTICAST] = data["image"]["multicast"]
 
         except KeyError as err:

supervisor/validate.py

@@ -26,6 +26,7 @@ from .const import (
     ATTR_LAST_BOOT,
     ATTR_LOGGING,
     ATTR_MULTICAST,
+    ATTR_OBSERVER,
     ATTR_PORT,
     ATTR_PORTS,
     ATTR_REFRESH_TOKEN,
@@ -141,6 +142,7 @@ SCHEMA_UPDATER_CONFIG = vol.Schema(
         vol.Optional(ATTR_CLI): vol.All(version_tag, str),
         vol.Optional(ATTR_DNS): vol.All(version_tag, str),
         vol.Optional(ATTR_AUDIO): vol.All(version_tag, str),
+        vol.Optional(ATTR_OBSERVER): vol.All(version_tag, str),
         vol.Optional(ATTR_MULTICAST): vol.All(version_tag, str),
         vol.Optional(ATTR_IMAGE, default=dict): vol.Schema(
             {
@@ -149,6 +151,7 @@ SCHEMA_UPDATER_CONFIG = vol.Schema(
                 vol.Optional(ATTR_CLI): docker_image,
                 vol.Optional(ATTR_DNS): docker_image,
                 vol.Optional(ATTR_AUDIO): docker_image,
+                vol.Optional(ATTR_OBSERVER): docker_image,
                 vol.Optional(ATTR_MULTICAST): docker_image,
             },
             extra=vol.REMOVE_EXTRA,