Merge pull request #487 from beer-garden/Subscriber_Types

Subscriber types

CHANGELOG.rst

@@ -1,6 +1,14 @@
 Brewtils Changelog
 ==================
 
+3.26.3
+------
+TBD
+
+- Added Subscriber Types to Subscriber model
+- Added Prefix Topics to System model
+- Support adding Prefix Topics for the Generated Subscribers. It is supported through the `@client` or SystemClient inputs or beer.conf
+
 3.26.2
 ------
 6/6/20

brewtils/decorators.py

@@ -40,6 +40,7 @@ def client(
     bg_version=None,  # type: Optional[str]
     group=None,  # type: str
     groups=[],  # type: Optional[List[str]]
+    prefix_topic=None,  # type: Optional[str]
 ):
     # type: (...) -> Type
     """Class decorator that marks a class as a beer-garden Client
@@ -68,14 +69,20 @@ def client(
         bg_version: Optional plugin version
         group: Optional plugin group
         groups: Optional plugin groups
+        prefix_topic: Optional prefix for Generated Command to Topic mappings
 
     Returns:
         The decorated class
 
     """
     if _wrapped is None:
         return functools.partial(
-            client, bg_name=bg_name, bg_version=bg_version, groups=groups, group=group
+            client,
+            bg_name=bg_name,
+            bg_version=bg_version,
+            groups=groups,
+            group=group,
+            prefix_topic=prefix_topic,
         )  # noqa
 
     # Assign these here so linters don't complain
@@ -84,6 +91,7 @@ def client(
     _wrapped._bg_commands = []
     _wrapped._current_request = None
     _wrapped._groups = groups
+    _wrapped._prefix_topic = prefix_topic
 
     if group:
         _wrapped._groups.append(group)

brewtils/models.py

@@ -760,6 +760,7 @@ def __init__(
         local=None,
         template=None,
         groups=None,
+        prefix_topic=None,
     ):
         self.name = name
         self.description = description
@@ -775,6 +776,7 @@ def __init__(
         self.local = local
         self.template = template
         self.groups = groups or []
+        self.prefix_topic = prefix_topic
 
     def __str__(self):
         return "%s:%s-%s" % (self.namespace, self.name, self.version)
@@ -1667,28 +1669,31 @@ def __init__(
         version=None,
         instance=None,
         command=None,
+        subscriber_type=None,
     ):
         self.garden = garden
         self.namespace = namespace
         self.system = system
         self.version = version
         self.instance = instance
         self.command = command
+        self.subscriber_type = subscriber_type or "DYNAMIC"
 
     def __str__(self):
         return "%s" % self.__dict__
 
     def __repr__(self):
         return (
             "<Subscriber: garden=%s, namespace=%s, system=%s, version=%s, instance=%s, "
-            "command=%s>"
+            "command=%s, subscriber_type=%s>"
             % (
                 self.garden,
                 self.namespace,
                 self.system,
                 self.version,
                 self.instance,
                 self.command,
+                self.subscriber_type,
             )
         )
 
@@ -1704,6 +1709,7 @@ def __eq__(self, other):
             and self.version == other.version
             and self.instance == other.instance
             and self.command == other.command
+            and self.subscriber_type == other.subscriber_type
         )
 
 

brewtils/plugin.py

@@ -165,6 +165,8 @@ class Plugin(object):
         group (str): Grouping label applied to plugin
         groups (list): Grouping labels applied to plugin
 
+        prefix_topic (str): Prefix for Generated Command Topics
+
         logger (:py:class:`logging.Logger`): Logger that will be used by the Plugin.
             Passing a logger will prevent the Plugin from preforming any additional
             logging configuration.
@@ -232,7 +234,7 @@ def __init__(self, client=None, system=None, logger=None, **kwargs):
 
         if not self._legacy:
             # Namespace setup depends on self._system and self._ez_client
-            self._setup_namespace()
+            self._setup_garden_namespace()
 
             # And with _system and _ez_client we can ask for the real logging config
             self._initialize_logging()
@@ -283,6 +285,10 @@ def _set_client(self, new_client):
             self._system.description = new_client.__doc__.split("\n")[0]
         if not self._system.groups:
             self._system.groups = getattr(new_client, "_groups", [])  # noqa
+        if not self.system.prefix_topic:
+            self._system.prefix_topic = getattr(
+                new_client, "_prefix_topic", None
+            )  # noqa
         # Now roll up / interpret all metadata to get the Commands
         self._system.commands = _parse_client(new_client)
 
@@ -298,6 +304,7 @@ def _set_client(self, new_client):
             client_clazz._bg_version = self._system.version
             client_clazz._bg_commands = self._system.commands
             client_clazz._groups = self._system.groups
+            client_clazz._prefix_topic = self._system.prefix_topic
             client_clazz._current_request = client_clazz.current_request
         except TypeError:
             if sys.version_info.major != 2:
@@ -368,7 +375,7 @@ def _startup(self):
 
         # If namespace couldn't be determined at init try one more time
         if not self._legacy and not self._config.namespace:
-            self._setup_namespace()
+            self._setup_garden_namespace()
 
         self._system = self._initialize_system()
         self._instance = self._initialize_instance()
@@ -747,7 +754,7 @@ def _setup_logging(self, logger=None, **kwargs):
 
         return logger or logging.getLogger(__name__)
 
-    def _setup_namespace(self):
+    def _setup_garden_namespace(self):
         """Determine the namespace the Plugin is operating in
 
         This function attempts to determine the correct namespace and ensures that
@@ -777,11 +784,16 @@ def _setup_namespace(self):
         before the namespace is determined will have potentially incorrect namespaces).
         """
         try:
-            ns = self._system.namespace or self._ez_client.get_config()["garden_name"]
+            garden = self._ez_client.get_config()["garden_name"]
+            ns = self._system.namespace or garden
 
             self._system.namespace = ns
             self._config.namespace = ns
+
+            self._config.garden = garden
+
             CONFIG.namespace = ns
+            CONFIG.garden = garden
         except Exception as ex:
             self._logger.warning(
                 "Namespace value was not resolved from config sources and an exception "
@@ -837,6 +849,7 @@ def _setup_system(self, system, plugin_kwargs):
                 display_name=self._config.display_name,
                 template=self._config.template,
                 groups=self._config.groups,
+                prefix_topic=self._config.prefix_topic,
             )
 
         return system

brewtils/rest/publish_client.py

@@ -88,13 +88,18 @@ def publish(
         """
 
         if _topic is None:
-            if (
-                brewtils.plugin.CONFIG.name
+
+            if brewtils.plugin._system.prefix_topic:
+                _topic = brewtils.plugin._system.prefix_topic
+            elif (
+                brewtils.plugin.CONFIG.garden
+                and brewtils.plugin.CONFIG.name
                 and brewtils.plugin.CONFIG.version
                 and brewtils.plugin.CONFIG.instance_name
                 and brewtils.plugin.CONFIG.namespace
             ):
-                _topic = "{0}.{1}.{2}.{3}".format(
+                _topic = "{0}.{1}.{2}.{3}.{4}".format(
+                    brewtils.plugin.CONFIG.garden,
                     brewtils.plugin.CONFIG.namespace,
                     brewtils.plugin.CONFIG.name,
                     brewtils.plugin.CONFIG.version,

brewtils/schemas.py

@@ -256,6 +256,7 @@ class SystemSchema(BaseSchema):
     local = fields.Bool(allow_none=True)
     template = fields.Str(allow_none=True)
     groups = fields.List(fields.Str(), allow_none=True)
+    prefix_topic = fields.Str(allow_none=True)
 
 
 class SystemDomainIdentifierSchema(BaseSchema):
@@ -633,6 +634,7 @@ class SubscriberSchema(BaseSchema):
     version = fields.Str(allow_none=True)
     instance = fields.Str(allow_none=True)
     command = fields.Str(allow_none=True)
+    subscriber_type = fields.Str(allow_none=True)
 
 
 class TopicSchema(BaseSchema):

brewtils/specification.py

@@ -162,6 +162,15 @@ def _is_json_dict(s):
         "required": False,
         "default": [],
     },
+    "prefix_topic": {
+        "type": "str",
+        "description": "Custom topic prefix to prepend to command name for Pub/Sub routing",
+        "required": False,
+        "long_description": "Each command has a generated topic assigned for Pub/Sub routing,"
+        "the default is "
+        "<garden name>.<namespace>.<system name>.<system version>.<system instance>.<command name>"
+        "if a prefix is provided, then it is `<prefix>.<command name>`",
+    },
 }
 
 _PLUGIN_SPEC = {

brewtils/test/fixtures.py

@@ -29,8 +29,8 @@
     RequestTemplate,
     Resolvable,
     Runner,
-    System,
     Subscriber,
+    System,
     Topic,
 )
 
@@ -262,6 +262,7 @@ def system_dict(instance_dict, command_dict, command_dict_2, system_id):
         "local": True,
         "template": "<html>template</html>",
         "groups": ["GroupB", "GroupA"],
+        "prefix_topic": "custom_topic",
     }
 
 
@@ -917,6 +918,7 @@ def subscriber_dict():
         "version": "1.0.0",
         "instance": "inst",
         "command": "run",
+        "subscriber_type": "DYNAMIC",
     }
 
 

test/decorators_test.py

@@ -511,9 +511,15 @@ def foo(self):
         assert hasattr(ClientClass, "_bg_commands")
         assert hasattr(ClientClass, "_current_request")
         assert hasattr(ClientClass, "_groups")
+        assert hasattr(ClientClass, "_prefix_topic")
 
     def test_with_args(self):
-        @client(bg_name="sys", bg_version="1.0.0", groups=["GroupA"])
+        @client(
+            bg_name="sys",
+            bg_version="1.0.0",
+            groups=["GroupA"],
+            prefix_topic="custom_topic",
+        )
         class ClientClass(object):
             @command
             def foo(self):
@@ -524,10 +530,12 @@ def foo(self):
         assert hasattr(ClientClass, "_bg_commands")
         assert hasattr(ClientClass, "_current_request")
         assert hasattr(ClientClass, "_groups")
+        assert hasattr(ClientClass, "_prefix_topic")
 
         assert ClientClass._bg_name == "sys"
         assert ClientClass._bg_version == "1.0.0"
         assert ClientClass._groups == ["GroupA"]
+        assert ClientClass._prefix_topic == "custom_topic"
 
     def test_group(self):
         @client(bg_name="sys", bg_version="1.0.0", group="GroupB")

test/plugin_test.py

@@ -54,6 +54,7 @@ def client():
         _bg_name=None,
         _bg_version=None,
         _groups=[],
+        _prefix_topic=None,
     )
 
 
@@ -150,6 +151,7 @@ def test_kwargs(self, client, bg_system):
             max_concurrent=1,
             group="GroupA",
             groups=["GroupB"],
+            prefix_topic="custom.topic",
         )
 
         assert plugin._logger == logger
@@ -158,6 +160,7 @@ def test_kwargs(self, client, bg_system):
         assert plugin._config.bg_url_prefix == "/beer/"
         assert plugin._config.ssl_enabled is False
         assert plugin._config.ca_verify is False
+        assert plugin._config.prefix_topic == "custom.topic"
         assert "GroupA" == plugin._config.group
         assert "GroupB" in plugin._config.groups
         assert "GroupA" not in plugin._config.groups
@@ -169,6 +172,7 @@ def test_env(self, client, bg_system):
         os.environ["BG_SSL_ENABLED"] = "False"
         os.environ["BG_CA_VERIFY"] = "False"
         os.environ["BG_GROUP"] = "GroupA"
+        os.environ["BG_PREFIX_TOPIC"] = "custom.topic"
 
         plugin = Plugin(client, system=bg_system, max_concurrent=1)
 
@@ -177,6 +181,7 @@ def test_env(self, client, bg_system):
         assert plugin._config.bg_url_prefix == "/beer/"
         assert plugin._config.ssl_enabled is False
         assert plugin._config.ca_verify is False
+        assert plugin._config.prefix_topic == "custom.topic"
         assert "GroupA" == plugin._config.group
 
     def test_conflicts(self, client, bg_system):
@@ -742,9 +747,10 @@ def test_from_config(self, client):
 
         self._validate_namespace(plugin, expected_namespace)
 
-    def test_from_system(self, client, bg_system):
+    def test_from_system(self, client, bg_system, ez_client):
         expected_namespace = "foo"
         bg_system.namespace = expected_namespace
+        ez_client.get_config.return_value = {"garden_name": "garden"}
 
         plugin = Plugin(client, bg_host="localhost", system=bg_system)
 

test/rest/publish_client_test.py

@@ -3,7 +3,7 @@
 
 import brewtils.rest
 from brewtils.errors import BrewtilsException
-from brewtils.models import Event, Events, Request
+from brewtils.models import Event, Events, Request, System
 from brewtils.rest.publish_client import PublishClient
 from brewtils.schema_parser import SchemaParser
 
@@ -25,8 +25,14 @@ def client():
     return PublishClient(bg_host="localhost", bg_port=3000)
 
 
+@pytest.fixture(autouse=True)
+def empty_system():
+    brewtils.plugin._system = System()
+
+
 class TestPublishClient(object):
     def setup_config(self):
+        brewtils.plugin.CONFIG.garden = "garden"
         brewtils.plugin.CONFIG.namespace = "foo"
         brewtils.plugin.CONFIG.name = "foo"
         brewtils.plugin.CONFIG.version = "1.0.0"
@@ -48,7 +54,17 @@ def test_missing_topic_found(self, client, easy_client):
         easy_client.publish_event.assert_called()
         called_event = easy_client.publish_event.call_args.args[0]
 
-        assert called_event.metadata["topic"] == "foo.foo.1.0.0.foo"
+        assert called_event.metadata["topic"] == "garden.foo.foo.1.0.0.foo"
+
+    def test_missing_prefix_topic_found(self, client, easy_client):
+        self.setup_config()
+        brewtils.plugin._system = System(prefix_topic="prefix.topic")
+        assert client.publish(no_topic="topic")
+
+        easy_client.publish_event.assert_called()
+        called_event = easy_client.publish_event.call_args.args[0]
+
+        assert called_event.metadata["topic"] == "prefix.topic"
 
     def test_verify_generated_request(self, client, easy_client):
         assert client.publish(