Allow tasks to notify a fqcn handler name (#68213)

* Allow tasks to notify a fqcn handler name

* Add tests. Fixes #68181

* Add changelog fragment

* Add test to ensure handlers are deduped properly with fqcn, role, and just handler names

* Add some docs about new special vars

11 files changed, 67 insertions, 18 deletions

diff --git a/changelogs/fragments/68181-fqcn-handler-notification.yml b/changelogs/fragments/68181-fqcn-handler-notification.yml
new file mode 100644
index 0000000000..a93d860157
--- /dev/null
+++ b/changelogs/fragments/68181-fqcn-handler-notification.yml
@@ -0,0 +1,6 @@
+bugfixes:
+- Allow tasks to notify a fqcn handler name (https://github.com/ansible/ansible/issues/68181)
+minor_changes:
+- Add new magic variable ``ansible_collection`` that contains the collection name
+- Add new magic variable ``ansible_role_name`` that contains the FQCN of the role
+- Updates ``ansible_role_names``, ``ansible_play_role_names``, and ``ansible_dependent_role_names`` to include the FQCN
diff --git a/docs/docsite/rst/reference_appendices/special_variables.rst b/docs/docsite/rst/reference_appendices/special_variables.rst
index 328f07a34d..fed8e9f8da 100644
--- a/docs/docsite/rst/reference_appendices/special_variables.rst
+++ b/docs/docsite/rst/reference_appendices/special_variables.rst
@@ -67,6 +67,12 @@ ansible_role_names
     The names of the roles currently imported into the current play, or roles referenced as dependencies of the roles
     imported into the current play.
 
+ansible_role_name
+    The fully qualified collection role name, in the format of ``namespace.collection.role_name``
+
+ansible_collection_name
+    The name of the collection the task that is executing is a part of. In the format of ``namespace.collection``
+
 ansible_run_tags
     Contents of the ``--tags`` CLI option, which specifies which tags will be included for the current run.
 
diff --git a/lib/ansible/playbook/role/__init__.py b/lib/ansible/playbook/role/__init__.py
index eb734f88c7..4b43b3eb64 100644
--- a/lib/ansible/playbook/role/__init__.py
+++ b/lib/ansible/playbook/role/__init__.py
@@ -128,7 +128,9 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
     def __repr__(self):
         return self.get_name()
 
-    def get_name(self):
+    def get_name(self, include_role_fqcn=True):
+        if include_role_fqcn:
+            return '.'.join(x for x in (self._role_collection, self._role_name) if x)
         return self._role_name
 
     @staticmethod
@@ -155,8 +157,8 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
             params['from_include'] = from_include
 
             hashed_params = hash_params(params)
-            if role_include.role in play.ROLE_CACHE:
-                for (entry, role_obj) in iteritems(play.ROLE_CACHE[role_include.role]):
+            if role_include.get_name() in play.ROLE_CACHE:
+                for (entry, role_obj) in iteritems(play.ROLE_CACHE[role_include.get_name()]):
                     if hashed_params == entry:
                         if parent_role:
                             role_obj.add_parent(parent_role)
@@ -169,11 +171,11 @@ class Role(Base, Conditional, Taggable, CollectionSearch):
             r = Role(play=play, from_files=from_files, from_include=from_include)
             r._load_role_data(role_include, parent_role=parent_role)
 
-            if role_include.role not in play.ROLE_CACHE:
-                play.ROLE_CACHE[role_include.role] = dict()
+            if role_include.get_name() not in play.ROLE_CACHE:
+                play.ROLE_CACHE[role_include.get_name()] = dict()
 
             # FIXME: how to handle cache keys for collection-based roles, since they're technically adjustable per task?
-            play.ROLE_CACHE[role_include.role][hashed_params] = r
+            play.ROLE_CACHE[role_include.get_name()][hashed_params] = r
             return r
 
         except RuntimeError:
diff --git a/lib/ansible/playbook/role/definition.py b/lib/ansible/playbook/role/definition.py
index 74b8334702..1f154c694c 100644
--- a/lib/ansible/playbook/role/definition.py
+++ b/lib/ansible/playbook/role/definition.py
@@ -231,3 +231,8 @@ class RoleDefinition(Base, Conditional, Taggable, CollectionSearch):
 
     def get_role_path(self):
         return self._role_path
+
+    def get_name(self, include_role_fqcn=True):
+        if include_role_fqcn:
+            return '.'.join(x for x in (self._role_collection, self.role) if x)
+        return self.role
diff --git a/lib/ansible/playbook/task.py b/lib/ansible/playbook/task.py
index e9e31a9a92..99e3b7a388 100644
--- a/lib/ansible/playbook/task.py
+++ b/lib/ansible/playbook/task.py
@@ -111,16 +111,19 @@ class Task(Base, Conditional, Taggable, CollectionSearch):
             path = "%s:%s" % (self._parent._play._ds._data_source, self._parent._play._ds._line_number)
         return path
 
-    def get_name(self):
+    def get_name(self, include_role_fqcn=True):
         ''' return the name of the task '''
 
-        if self._role and self.name and ("%s : " % self._role._role_name) not in self.name:
-            return "%s : %s" % (self._role.get_name(), self.name)
+        if self._role:
+            role_name = self._role.get_name(include_role_fqcn=include_role_fqcn)
+
+        if self._role and self.name and role_name not in self.name:
+            return "%s : %s" % (role_name, self.name)
         elif self.name:
             return self.name
         else:
             if self._role:
-                return "%s : %s" % (self._role.get_name(), self.action)
+                return "%s : %s" % (role_name, self.action)
             else:
                 return "%s" % (self.action,)
 
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index c4a46dfcb2..63cb0b18a9 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -460,11 +460,14 @@ class StrategyBase:
                             # include the role name (if the handler is from a role). If that
                             # is not found, we resort to the simple name field, which doesn't
                             # have anything extra added to it.
-                            if handler_task.name == handler_name:
+                            candidates = (
+                                handler_task.name,
+                                handler_task.get_name(include_role_fqcn=False),
+                                handler_task.get_name(include_role_fqcn=True),
+                            )
+
+                            if handler_name in candidates:
                                 return handler_task
-                            else:
-                                if handler_task.get_name() == handler_name:
-                                    return handler_task
                         except (UndefinedError, AnsibleUndefinedVariable):
                             # We skip this handler due to the fact that it may be using
                             # a variable in the name that was conditionally included via
@@ -705,7 +708,7 @@ class StrategyBase:
             if original_task._role is not None and role_ran:  # TODO:  and original_task.action != 'include_role':?
                 # lookup the role in the ROLE_CACHE to make sure we're dealing
                 # with the correct object and mark it as executed
-                for (entry, role_obj) in iteritems(iterator._play.ROLE_CACHE[original_task._role._role_name]):
+                for (entry, role_obj) in iteritems(iterator._play.ROLE_CACHE[original_task._role.get_name()]):
                     if role_obj._uuid == original_task._role._uuid:
                         role_obj._had_task_run[original_host.name] = True
 
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index 18925d8703..143f61893f 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -457,9 +457,9 @@ class VariableManager:
 
         if play:
             # This is a list of all role names of all dependencies for all roles for this play
-            dependency_role_names = list(set([d._role_name for r in play.roles for d in r.get_all_dependencies()]))
+            dependency_role_names = list(set([d.get_name() for r in play.roles for d in r.get_all_dependencies()]))
             # This is a list of all role names of all roles for this play
-            play_role_names = [r._role_name for r in play.roles]
+            play_role_names = [r.get_name() for r in play.roles]
 
             # ansible_role_names includes all role names, dependent or directly referenced by the play
             variables['ansible_role_names'] = list(set(dependency_role_names + play_role_names))
@@ -477,9 +477,11 @@ class VariableManager:
 
         if task:
             if task._role:
-                variables['role_name'] = task._role.get_name()
+                variables['role_name'] = task._role.get_name(include_role_fqcn=False)
                 variables['role_path'] = task._role._role_path
                 variables['role_uuid'] = text_type(task._role._uuid)
+                variables['ansible_collection_name'] = task._role._role_collection
+                variables['ansible_role_name'] = task._role.get_name()
 
         if self._inventory is not None:
             variables['groups'] = self._inventory.get_groups_dict()
diff --git a/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/common_handlers/handlers/main.yml b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/common_handlers/handlers/main.yml
new file mode 100644
index 0000000000..d9f732317e
--- /dev/null
+++ b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/common_handlers/handlers/main.yml
@@ -0,0 +1,6 @@
+# This handler should only be called 1 time, if it's called more than once
+# this task should fail on subsequent executions
+- name: test_fqcn_handler
+  set_fact:
+    handler_counter: '{{ handler_counter|int + 1 }}'
+  failed_when: handler_counter|int > 1
diff --git a/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/meta/main.yml b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/meta/main.yml
new file mode 100644
index 0000000000..9218f3d7c9
--- /dev/null
+++ b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/meta/main.yml
@@ -0,0 +1,2 @@
+dependencies:
+  - testns.testcoll.common_handlers
diff --git a/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/tasks/main.yml b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/tasks/main.yml
new file mode 100644
index 0000000000..db8767d217
--- /dev/null
+++ b/test/integration/targets/collections/collection_root_user/ansible_collections/testns/testcoll/roles/test_fqcn_handlers/tasks/main.yml
@@ -0,0 +1,7 @@
+- debug:
+    msg: Fire fqcn handler
+  changed_when: true
+  notify:
+    - 'testns.testcoll.common_handlers : test_fqcn_handler'
+    - 'common_handlers : test_fqcn_handler'
+    - 'test_fqcn_handler'
diff --git a/test/integration/targets/collections/posix.yml b/test/integration/targets/collections/posix.yml
index 89e2a654bf..d1ea2f9c04 100644
--- a/test/integration/targets/collections/posix.yml
+++ b/test/integration/targets/collections/posix.yml
@@ -394,3 +394,10 @@
       that:
       - hostvars['dynamic_host_a'] is defined
       - hostvars['dynamic_host_a'].connection_out.stdout == "localconn ran echo 'hello world'"
+
+- name: Test FQCN handlers
+  hosts: testhost
+  vars:
+    handler_counter: 0
+  roles:
+    - testns.testcoll.test_fqcn_handlers