refactor: Make Discovery a DataClass.

And move it out of the fsm.py file.

Signed-off-by: Paulo Vital <[email protected]>

Author: pvital

src/instana/agent/host.py

@@ -9,22 +9,25 @@
 import json
 import os
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 import requests
 import urllib3
 from requests import Response
 
 from instana.agent.base import BaseAgent
 from instana.collector.host import HostCollector
-from instana.fsm import Discovery, TheMachine
+from instana.fsm import TheMachine
 from instana.log import logger
 from instana.options import StandardOptions
 from instana.util import to_json
 from instana.util.runtime import get_py_source, log_runtime_env_info
 from instana.util.span_utils import get_operation_specifiers
 from instana.version import VERSION
 
+if TYPE_CHECKING:
+    from instana.util.process_discovery import Discovery
+
 
 class AnnounceData(object):
     """The Announce Payload"""
@@ -176,7 +179,7 @@ def is_agent_listening(
 
     def announce(
         self,
-        discovery: Discovery,
+        discovery: "Discovery",
     ) -> Optional[Dict[str, Any]]:
         """
         With the passed in Discovery class, attempt to announce to the host agent.

src/instana/fsm.py

@@ -8,46 +8,23 @@
 import subprocess
 import sys
 import threading
-from typing import TYPE_CHECKING, Any, Callable, Optional
+from typing import TYPE_CHECKING, Any, Callable
 
 from fysom import Fysom
 
 from instana.log import logger
 from instana.util import get_default_gateway
+from instana.util.process_discovery import Discovery
 from instana.version import VERSION
 
 if TYPE_CHECKING:
     from instana.agent.host import HostAgent
 
 
-class Discovery:
-    pid: int = 0
-    name: Optional[str] = None
-    args: Optional[List[str]] = None
-    fd: int = -1
-    inode: str = ""
-
-    def __init__(self, **kwds: Any) -> None:
-        self.__dict__.update(kwds)
-
-    def to_dict(self) -> Dict[str, Any]:
-        kvs: Dict[str, Any] = dict()
-        kvs["pid"] = self.pid
-        kvs["name"] = self.name
-        kvs["args"] = self.args
-        kvs["fd"] = self.fd
-        kvs["inode"] = self.inode
-        return kvs
-
-
 class TheMachine:
     RETRY_PERIOD = 30
     THREAD_NAME = "Instana Machine"
 
-    agent: Optional["HostAgent"] = None
-    fsm = None
-    timer = None
-
     warnedPeriodic = False
 
     def __init__(self, agent: "HostAgent") -> None:

src/instana/util/process_discovery.py

@@ -0,0 +1,13 @@
+# (c) Copyright IBM Corp. 2025
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+
+@dataclass
+class Discovery:
+    pid: int = 0  # the PID of this process
+    name: Optional[str] = None  # the name of the executable
+    args: Optional[List[str]] = None  # the command line arguments
+    fd: int = -1  # the file descriptor of the socket associated with the connection to the agent for this HTTP request
+    inode: str = ""  # the inode of the socket associated with the connection to the agent for this HTTP request

tests/agent/test_host.py

@@ -14,12 +14,13 @@
 
 from instana.agent.host import AnnounceData, HostAgent
 from instana.collector.host import HostCollector
-from instana.fsm import Discovery, TheMachine
+from instana.fsm import TheMachine
 from instana.options import StandardOptions
 from instana.recorder import StanRecorder
 from instana.singletons import get_agent
 from instana.span.span import InstanaSpan
 from instana.span_context import SpanContext
+from instana.util.process_discovery import Discovery
 from instana.util.runtime import is_windows
 
 
@@ -715,3 +716,4 @@ def test_is_service_or_endpoint_ignored(self) -> None:
 
         # don't ignore other services
         assert not self.agent._HostAgent__is_endpoint_ignored("service3")
+        assert not self.agent._HostAgent__is_endpoint_ignored("service3")