Issue/6025 implement discovery handler

src/inmanta/agent/handler.py

@@ -22,18 +22,19 @@
 import typing
 import uuid
 from abc import ABC, abstractmethod
-from collections import defaultdict
+from collections import abc, defaultdict
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, Generic, List, Optional, Tuple, Type, TypeVar, Union, cast, overload
 
+import pydantic
 from tornado import concurrent
 
 import inmanta
 from inmanta import const, data, protocol, resources
 from inmanta.agent import io
 from inmanta.agent.cache import AgentCache
 from inmanta.const import ParameterSource, ResourceState
-from inmanta.data.model import AttributeStateChange, ResourceIdStr
+from inmanta.data.model import AttributeStateChange, DiscoveredResource, ResourceIdStr
 from inmanta.protocol import Result, json_encode
 from inmanta.stable_api import stable_api
 from inmanta.types import SimpleTypes
@@ -47,6 +48,10 @@
 LOGGER = logging.getLogger(__name__)
 
 T = TypeVar("T")
+# Discovery Resource Type
+DRT = TypeVar("DRT", bound=resources.Resource)
+# Unmanaged Resource Type
+URT = TypeVar("URT", bound=pydantic.BaseModel)  # bound=... enough to enforce the fact it needs to be serializable ?
 T_FUNC = TypeVar("T_FUNC", bound=Callable[..., Any])
 
 
@@ -55,7 +60,7 @@ class provider(object):  # noqa: N801
     """
     A decorator that registers a new handler.
 
-    :param resource_type: The type of the resource this handler provides an implementation for.
+    :param resource_type: The type of the resource this handler is responsible for.
                           For example, :inmanta:entity:`std::File`
     :param name: A name to reference this provider.
     """
@@ -419,34 +424,45 @@ def _log_msg(self, level: int, msg: str, *args: object, exc_info: bool = False,
 
 
 @stable_api
-class ResourceHandler(object):
+class HandlerABC(ABC):
+    """
+    Top-level abstract base class all handlers should inherit from.
     """
-    A baseclass for classes that handle resources. New handler are registered with the
-    :func:`~inmanta.agent.handler.provider` decorator.
 
-    The implementation of a handler should use the ``self._io`` instance to execute io operations. This io objects
-    makes abstraction of local or remote operations. See :class:`~inmanta.agent.io.local.LocalIO` for the available
-    operations.
+    # _client: Optional[protocol.SessionClient] = None
+    # _agent: Optional[inmanta.agent.agent.AgentInstance] = None
+    # _io_loop = None
 
-    :param agent: The agent that is executing this handler.
-    :param io: The io object to use.
-    """
+    def pre(self, ctx: HandlerContext, resource: resources.Resource) -> None:
+        """
+        Method executed before a handler operation (Facts, dryrun, real deployment, ...) is executed. Override this method
+        to run before an operation.
 
-    def __init__(self, agent: "inmanta.agent.agent.AgentInstance", io: Optional["IOBase"] = None) -> None:
-        self._agent = agent
+        :param ctx: Context object to report changes and logs to the agent and server.
+        :param resource: The resource to query facts for.
+        """
 
-        if io is None:
-            raise Exception("Unsupported: no resource mgmt in RH")
-        else:
-            self._io = io
+    def post(self, ctx: HandlerContext, resource: resources.Resource) -> None:
+        """
+        Method executed after an operation. Override this method to run after an operation.
 
-        self._client: Optional[protocol.SessionClient] = None
-        # explicit ioloop reference, as we don't want the ioloop for the current thread, but the one for the agent
-        self._ioloop = agent.process._io_loop
+        :param ctx: Context object to report changes and logs to the agent and server.
+        :param resource: The resource to query facts for.
+        """
+
+    def get_client(self) -> protocol.SessionClient:
+        """
+        Get the client instance that identifies itself with the agent session.
+
+        :return: A client that is associated with the session of the agent that executes this handler.
+        """
+        if self._client is None:
+            self._client = protocol.SessionClient("agent", self._agent.sessionid)
+        return self._client
 
     def run_sync(self, func: typing.Callable[[], typing.Awaitable[T]]) -> T:
         """
-        Run a the given async function on the ioloop of the agent. It will block the current thread until the future
+        Run the given async function on the ioloop of the agent. It will block the current thread until the future
         resolves.
 
         :param func: A function that returns a yieldable future.
@@ -470,18 +486,35 @@ def run() -> None:
 
         return f.result()
 
-    def set_cache(self, cache: AgentCache) -> None:
-        self.cache = cache
 
-    def get_client(self) -> protocol.SessionClient:
-        """
-        Get the client instance that identifies itself with the agent session.
+@stable_api
+class ResourceHandler(HandlerABC):
+    """
+    A baseclass for classes that handle resources. New handler are registered with the
+    :func:`~inmanta.agent.handler.provider` decorator.
 
-        :return: A client that is associated with the session of the agent that executes this handler.
-        """
-        if self._client is None:
-            self._client = protocol.SessionClient("agent", self._agent.sessionid)
-        return self._client
+    The implementation of a handler should use the ``self._io`` instance to execute io operations. This io objects
+    makes abstraction of local or remote operations. See :class:`~inmanta.agent.io.local.LocalIO` for the available
+    operations.
+
+    :param agent: The agent that is executing this handler.
+    :param io: The io object to use.
+    """
+
+    def __init__(self, agent: "inmanta.agent.agent.AgentInstance", io: Optional["IOBase"] = None) -> None:
+        self._agent = agent
+
+        if io is None:
+            raise Exception("Unsupported: no resource mgmt in RH")
+        else:
+            self._io = io
+
+        self._client: Optional[protocol.SessionClient] = None
+        # explicit ioloop reference, as we don't want the ioloop for the current thread, but the one for the agent
+        self._ioloop = agent.process._io_loop
+
+    def set_cache(self, cache: AgentCache) -> None:
+        self.cache = cache
 
     def can_reload(self) -> bool:
         """
@@ -499,23 +532,6 @@ def do_reload(self, ctx: HandlerContext, resource: resources.Resource) -> None:
         :param resource: The resource to reload.
         """
 
-    def pre(self, ctx: HandlerContext, resource: resources.Resource) -> None:
-        """
-        Method executed before a handler operation (Facts, dryrun, real deployment, ...) is executed. Override this method
-        to run before an operation.
-
-        :param ctx: Context object to report changes and logs to the agent and server.
-        :param resource: The resource to query facts for.
-        """
-
-    def post(self, ctx: HandlerContext, resource: resources.Resource) -> None:
-        """
-        Method executed after an operation. Override this method to run after an operation.
-
-        :param ctx: Context object to report changes and logs to the agent and server.
-        :param resource: The resource to query facts for.
-        """
-
     def close(self) -> None:
         pass
 
@@ -582,7 +598,7 @@ def deploy(
         requires: Dict[ResourceIdStr, ResourceState],
     ) -> None:
         """
-        This method is always be called by the agent, even when one of the requires of the given resource
+        This method is always called by the agent, even when one of the requires of the given resource
         failed to deploy. The default implementation of this method will deploy the given resource when all its
         requires were deployed successfully. Override this method if a different condition determines whether the
         resource should deploy.
@@ -961,6 +977,109 @@ def execute(self, ctx: HandlerContext, resource: TPurgeableResource, dry_run: bo
         super().execute(ctx, resource, dry_run)
 
 
+@stable_api
+class DiscoveryHandler(HandlerABC, Generic[DRT, URT]):
+    # The DiscoveryHandler is generic in both the handler's Discovery Resource type (DRT)
+    # and the Unmanaged Resource type (URT) it reports to the server. The second has to be serializable.
+
+    # This class deploys instances of DRT and reports URT to the server.
+
+    def __init__(self, agent: "inmanta.agent.agent.AgentInstance") -> None:
+        self._agent = agent
+        self._ioloop = agent._io_loop
+
+    @abstractmethod
+    def discover_resources(self, ctx: HandlerContext, discovery_resource: DRT) -> abc.Mapping[ResourceIdStr, URT]:
+        raise NotImplementedError
+
+    def report_discovered_resources(self, ctx: HandlerContext, resource: DRT) -> None:
+        """ """
+
+        def _call_discovered_resource_create_batch() -> typing.Awaitable[Result]:
+            return self.get_client().discovered_resource_create_batch(
+                tid=self._agent.environment, discovered_resources=discovered_resources
+            )
+
+        try:
+            self.pre(ctx, resource)
+            # report to the server
+            discovered_resources: List[DiscoveredResource] = [
+                DiscoveredResource(discovered_resource_id=resource_id, values=values)
+                for resource_id, values in self.discover_resources(ctx, resource).items()
+            ]
+            result = self.run_sync(_call_discovered_resource_create_batch)
+
+            if result.code != 200:
+                error_msg_from_server = f": {result.result['message']}" if "message" in result.result else ""
+                raise Exception(f"Failed to report discovered resources to the server{error_msg_from_server}")
+
+        except Exception as e:
+            ctx.set_status(const.ResourceState.failed)
+            ctx.exception(
+                (
+                    "An error occurred during resource discovery of type %(urt)s, "
+                    "triggered by %(resource_id)s (exception: %(exception)s"
+                ),
+                urt=str(URT),
+                resource_id=resource.id,
+                exception=f"{e.__class__.__name__}('{e}')",
+            )
+        finally:
+            try:
+                self.post(ctx, resource)
+            except Exception as e:
+                ctx.exception(
+                    (
+                        "An error occurred after resource discovery of type %(urt)s, "
+                        "triggered by %(resource_id)s (exception: %(exception)s"
+                    ),
+                    urt=str(URT),
+                    resource_id=resource.id,
+                    exception=f"{e.__class__.__name__}('{e}')",
+                )
+
+    async def async_report_discovered_resources(self, ctx: HandlerContext, resource: DRT) -> None:
+        try:
+            self.pre(ctx, resource)
+            # report to the server
+            discovered_resources: List[DiscoveredResource] = [
+                DiscoveredResource(discovered_resource_id=resource_id, values=values)
+                for resource_id, values in self.discover_resources(ctx, resource).items()
+            ]
+            result = await self.get_client().discovered_resource_create_batch(
+                tid=self._agent.environment, discovered_resources=discovered_resources
+            )
+
+            if result.code != 200:
+                error_msg_from_server = f": {result.result['message']}" if "message" in result.result else ""
+                raise Exception(f"Failed to report discovered resources to the server{error_msg_from_server}")
+
+        except Exception as e:
+            ctx.set_status(const.ResourceState.failed)
+            ctx.exception(
+                (
+                    "An error occurred during resource discovery of type %(urt)s, "
+                    "triggered by %(resource_id)s (exception: %(exception)s"
+                ),
+                urt=str(URT),
+                resource_id=resource.id,
+                exception=f"{e.__class__.__name__}('{e}')",
+            )
+        finally:
+            try:
+                self.post(ctx, resource)
+            except Exception as e:
+                ctx.exception(
+                    (
+                        "An error occurred after resource discovery of type %(urt)s, "
+                        "triggered by %(resource_id)s (exception: %(exception)s"
+                    ),
+                    urt=str(URT),
+                    resource_id=resource.id,
+                    exception=f"{e.__class__.__name__}('{e}')",
+                )
+
+
 class Commander(object):
     """
     This class handles commands

src/inmanta/resources.py

@@ -464,6 +464,15 @@ class PurgeableResource(Resource):
     purge_on_delete: bool
 
 
+@stable_api
+class DiscoveryResource(Resource):
+    """
+    See :inmanta:entity:`std::DiscoveryResource` for more information.
+    """
+
+    fields = ()
+
+
 @stable_api
 class ManagedResource(Resource):
     """
@@ -499,6 +508,13 @@ class Id(object):
     """
 
     def __init__(self, entity_type: str, agent_name: str, attribute: str, attribute_value: str, version: int = 0) -> None:
+        """
+        :attr entity_type: The resource type, as defined in the configuration model. For example :inmanta:entity:`std::File`.
+        :attr agent_name: The agent responsible for this resource.
+        :attr attribute: The key attribute that uniquely identifies this resource on the agent
+        :attr attribute_value: The corresponding value for this key attribute.
+        :attr version: The version number for this resource.
+        """
         self._entity_type = entity_type
         self._agent_name = agent_name
         self._attribute = attribute
@@ -553,6 +569,18 @@ def __eq__(self, other: object) -> bool:
         return str(self) == str(other) and type(self) == type(other)
 
     def resource_str(self) -> ResourceIdStr:
+        """
+        String representation for this resource with the following format:
+
+            <type>[<agent>,<attribute>=<value>]
+
+            - type: The resource type, as defined in the configuration model. For example :inmanta:entity:`std::File`.
+            - agent: The agent responsible for this resource.
+            - attribute: The key attribute that uniquely identifies this resource on the agent
+            - value: The corresponding value for this key attribute.
+
+        :return: Returns a :py:class:`inmanta.data.model.ResourceIdStr`
+        """
         return cast(
             ResourceIdStr,
             "%(type)s[%(agent)s,%(attribute)s=%(value)s]"