Route hosts to instance groups from an inventory variable

In deployments with several isolated network zones, each zone has its
own execution nodes grouped in an instance group, because firewalls only
let those nodes reach the hosts of their zone. A job runs on a single
instance group, so the same baseline template (sysctl, users, patching)
has to be duplicated per zone with a different limit and instance group.

This adds an instance_group_routing_var field to job templates: the name
of an inventory variable (usually a group var) whose value names the
instance group each host should run on. At launch, hosts are grouped by
the resolved value and the launch creates the same implicit workflow job
that sliced jobs use, with one node per bucket: each job runs restricted
to its bucket's hosts and assigned to the routed instance group. Hosts
that do not resolve the variable run in an extra job with the normal
instance group selection, and if everything lands in one bucket a plain
job is launched on it. Resolution follows a simplified version of the
Ansible precedence rules (host vars win, then groups by depth, priority
and name) and only considers enabled hosts.

Instance groups are protected by RBAC and the routing variable is
inventory data, so it is not trusted blindly: the launch endpoint checks
the launching user's use permission on every referenced instance group,
the same rule already applied to instance groups prompted at launch, and
unknown names reject the launch instead of falling back silently. The
validated buckets are handed to the job creation so the launch routes to
exactly what was checked. The same checks run again when relaunching a
routed workflow, which recomputes the buckets from current inventory
data. Routing is mutually exclusive with job slicing (also when the
slice count is prompted at launch), prompting instance groups overrides
it, and provisioning callbacks never route, like they never slice.

Launch paths without a requesting user stay loud but safe when a routing
value names a missing instance group: a workflow containing the routed
template is marked failed with the reason instead of crashing the
workflow manager cycle, webhook launches answer 400, and relaunching a
routed child job keeps targeting its original bucket even if the
template's routing variable changed since.

Refs #566

Author: fernandorocagonzalez

awx/api/serializers.py

@@ -3180,6 +3180,7 @@ class Meta:
             'scm_branch',
             'forks',
             'limit',
+            'instance_group_routing_var',
             'verbosity',
             'extra_vars',
             'job_tags',
@@ -3385,6 +3386,13 @@ def get_field_from_model_or_attrs(fd):
         elif inventory is None and not get_field_from_model_or_attrs('ask_inventory_on_launch'):
             raise serializers.ValidationError({'inventory': prompting_error_message})
 
+        instance_group_routing_var = get_field_from_model_or_attrs('instance_group_routing_var')
+        if instance_group_routing_var:
+            if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', instance_group_routing_var):
+                raise serializers.ValidationError({'instance_group_routing_var': _("Must be a valid variable name.")})
+            if (get_field_from_model_or_attrs('job_slice_count') or 1) > 1:
+                raise serializers.ValidationError({'instance_group_routing_var': _("Instance group routing cannot be combined with job slicing.")})
+
         return super(JobTemplateSerializer, self).validate(attrs)
 
     def validate_extra_vars(self, value):
@@ -3435,6 +3443,7 @@ class Meta:
             'diff_mode',
             'job_slice_number',
             'job_slice_count',
+            'instance_group_routing_value',
             'webhook_service',
             'webhook_credential',
             'webhook_guid',

awx/api/views/__init__.py

@@ -2435,6 +2435,14 @@ def post(self, request, *args, **kwargs):
         if not request.user.can_access(models.JobLaunchConfig, 'add', serializer.validated_data, template=obj):
             raise PermissionDenied()
 
+        ig_routing_error, ig_routing_buckets = obj.get_ig_routing_launch_error(request.user, serializer.validated_data)
+        if ig_routing_error:
+            return Response(dict(errors=[ig_routing_error]), status=status.HTTP_400_BAD_REQUEST)
+        if ig_routing_buckets is not None:
+            # hand the validated buckets to create_unified_job, so the launch
+            # routes to exactly the instance groups that passed the check
+            serializer.validated_data['_ig_routing_buckets'] = ig_routing_buckets
+
         passwords = serializer.validated_data.pop('credential_passwords', {})
         new_job = obj.create_unified_job(**serializer.validated_data)
         result = new_job.signal_start(**passwords)
@@ -2856,6 +2864,7 @@ def post(self, request, *args, **kwargs):
             extra_vars_redacted, removed = extract_ansible_vars(extra_vars)
             kv['extra_vars'] = extra_vars_redacted
         kv['_prevent_slicing'] = True  # will only run against 1 host, so no point
+        kv['_prevent_ig_routing'] = True  # same reason: a single host needs no routing fan-out
         with transaction.atomic():
             job = job_template.create_job(**kv)
 
@@ -3223,6 +3232,34 @@ def post(self, request, *args, **kwargs):
             jt = obj.job_template
             if not jt:
                 raise ParseError(_('Cannot relaunch slice workflow job orphaned from job template.'))
+            # classify by what actually spawned the workflow, not by the current
+            # (editable) template configuration
+            is_ig_routed_workflow = any('ig_routing_value' in node.ancestor_artifacts for node in obj.workflow_nodes.all())
+            # the relaunch re-applies the original prompts, so it targets the
+            # prompted inventory when the template asks for one, or whatever
+            # inventory the template has now otherwise
+            relaunch_inventory = obj.inventory if (jt.ask_inventory_on_launch and obj.inventory) else jt.inventory
+            if is_ig_routed_workflow:
+                if not jt.instance_group_routing_var:
+                    raise ParseError(
+                        _('Cannot relaunch instance group routed workflow job: the job template no longer routes by variable. Launch the job template instead.')
+                    )
+                if relaunch_inventory is None:
+                    raise ParseError(_('Cannot relaunch instance group routed workflow job without an inventory.'))
+                if getattr(relaunch_inventory, 'kind', None) != 'federated':
+                    # routing buckets are recomputed at relaunch, so only require
+                    # that the inventory still fans out and that the relaunching
+                    # user may use the routed instance groups
+                    relaunch_kwargs = {'inventory': relaunch_inventory} if jt.ask_inventory_on_launch else {}
+                    ig_routing_error, ig_routing_buckets = jt.get_ig_routing_launch_error(request.user, relaunch_kwargs)
+                    if ig_routing_error:
+                        raise ParseError(ig_routing_error)
+                    if ig_routing_buckets is None or len(ig_routing_buckets) <= 1:
+                        raise ParseError(
+                            _(
+                                'Cannot relaunch instance group routed workflow job: the inventory no longer routes to multiple instance groups. Launch the job template instead.'
+                            )
+                        )
             elif getattr(obj.inventory, 'kind', None) != 'federated' and (
                 not obj.inventory or min(obj.inventory.hosts.count(), jt.job_slice_count) != obj.workflow_nodes.count()
             ):

awx/api/views/webhooks.py

@@ -3,6 +3,7 @@
 import logging
 import urllib.parse
 
+from django.core.exceptions import ValidationError as DjangoValidationError
 from django.utils.encoding import force_bytes
 from django.utils.translation import gettext_lazy as _
 from django.views.decorators.csrf import csrf_exempt
@@ -167,7 +168,12 @@ def post(self, request, *args, **kwargs_in):
             kwargs['extra_vars']['{}_webhook_status_api'.format(name)] = status_api
             kwargs['extra_vars']['{}_webhook_payload'.format(name)] = request.data
 
-        new_job = obj.create_unified_job(**kwargs)
+        try:
+            new_job = obj.create_unified_job(**kwargs)
+        except DjangoValidationError as exc:
+            # e.g. instance group routing pointing at an instance group that does
+            # not exist; answer with a 4xx instead of letting the SCM see a 500
+            return Response(dict(errors=exc.messages), status=status.HTTP_400_BAD_REQUEST)
         new_job.signal_start()
 
         return Response({'message': "Job queued."}, status=status.HTTP_202_ACCEPTED)

awx/main/migrations/0201_instance_group_routing.py

@@ -0,0 +1,43 @@
+# Generated by Django 5.2.15 on 2026-07-05 12:00
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0200_add_list_ordering'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='job',
+            name='instance_group_routing_var',
+            field=models.CharField(
+                blank=True,
+                default='',
+                help_text='Name of an inventory variable used to route hosts to instance groups at launch. When set, hosts are grouped by the value of this variable (host variables win over group variables) and the launch spawns one job per group, restricted to its hosts and assigned to the instance group named by the value. Hosts without the variable run in an extra job with the normal instance group selection. Launch fails if a value does not name an existing instance group or the launching user lacks use permission on it. Cannot be combined with job slicing.',
+                max_length=1024,
+            ),
+        ),
+        migrations.AddField(
+            model_name='jobtemplate',
+            name='instance_group_routing_var',
+            field=models.CharField(
+                blank=True,
+                default='',
+                help_text='Name of an inventory variable used to route hosts to instance groups at launch. When set, hosts are grouped by the value of this variable (host variables win over group variables) and the launch spawns one job per group, restricted to its hosts and assigned to the instance group named by the value. Hosts without the variable run in an extra job with the normal instance group selection. Launch fails if a value does not name an existing instance group or the launching user lacks use permission on it. Cannot be combined with job slicing.',
+                max_length=1024,
+            ),
+        ),
+        migrations.AddField(
+            model_name='job',
+            name='instance_group_routing_value',
+            field=models.TextField(
+                default=None,
+                editable=False,
+                help_text="If created by instance group routing, the routing variable value that selected this job's bucket of hosts. An empty string means the bucket of hosts that do not resolve the variable. Null when the job was not routed.",
+                null=True,
+            ),
+        ),
+    ]

awx/main/models/inventory.py

@@ -300,7 +300,101 @@ def get_sliced_hosts(self, host_queryset, slice_number, slice_count):
             host_queryset = host_queryset[offset::slice_count]
         return host_queryset
 
-    def get_script_data(self, hostvars=False, towervars=False, show_all=False, slice_number=1, slice_count=1):
+    def resolve_host_variable(self, var_name):
+        """
+        Resolve the value of an inventory variable for every enabled host,
+        following a simplified version of the Ansible precedence rules: host
+        variables win over group variables, and groups are merged sorted by
+        depth, then ansible_group_priority, then name, later ones winning.
+
+        Returns a dict mapping host name to value, containing only the hosts
+        where the variable resolves to a non empty string; hosts where it is
+        unset, empty or not a string are left out. Disabled hosts are ignored,
+        matching the hosts a job actually runs against.
+        """
+        group_sort_keys = {}
+        group_raw_values = {}
+        for group in self.groups.only('id', 'name', 'variables'):
+            group_vars = group.variables_dict
+            try:
+                priority = int(group_vars.get('ansible_group_priority', 1))
+            except (TypeError, ValueError):
+                priority = 1
+            group_sort_keys[group.id] = (priority, group.name)
+            if var_name in group_vars:
+                group_raw_values[group.id] = group_vars[var_name]
+
+        parent_map = self.get_group_parents_map() if group_sort_keys else {}
+
+        depths = {}
+
+        def group_depth(group_id, visiting):
+            if group_id in depths:
+                return depths[group_id]
+            depth = 1
+            for parent_id in parent_map.get(group_id, ()):
+                if parent_id in visiting:
+                    continue  # the API prevents group cycles, guard just in case
+                depth = max(depth, group_depth(parent_id, visiting | {parent_id}) + 1)
+            depths[group_id] = depth
+            return depth
+
+        ancestor_map = {}
+
+        def group_ancestors(group_id, visiting):
+            if group_id in ancestor_map:
+                return ancestor_map[group_id]
+            found = set()
+            for parent_id in parent_map.get(group_id, ()):
+                if parent_id in visiting:
+                    continue
+                found.add(parent_id)
+                found |= group_ancestors(parent_id, visiting | {parent_id})
+            ancestor_map[group_id] = found
+            return found
+
+        host_group_map = {}
+        if group_sort_keys:
+            for group_id, host_ids in self.get_group_hosts_map().items():
+                for host_id in host_ids:
+                    host_group_map.setdefault(host_id, set()).add(group_id)
+
+        result = {}
+        for host in self.hosts.filter(enabled=True).only('id', 'name', 'variables'):
+            host_vars = host.variables_dict
+            if var_name in host_vars:
+                value = host_vars[var_name]
+            else:
+                value = None
+                candidate_ids = set()
+                for group_id in host_group_map.get(host.id, ()):
+                    candidate_ids.add(group_id)
+                    candidate_ids |= group_ancestors(group_id, {group_id})
+                best_key = None
+                for group_id in candidate_ids:
+                    if group_id not in group_raw_values:
+                        continue
+                    priority, group_name = group_sort_keys[group_id]
+                    key = (group_depth(group_id, {group_id}), priority, group_name)
+                    if best_key is None or key > best_key:
+                        best_key = key
+                        value = group_raw_values[group_id]
+            if isinstance(value, str) and value:
+                result[host.name] = value
+        return result
+
+    def filter_hosts_to_routing_bucket(self, hosts, var_name, value):
+        """
+        Restrict hosts to the instance group routing bucket for the given value:
+        the hosts whose routing variable resolves to it, or the hosts that do
+        not resolve the variable at all when the value is an empty string.
+        """
+        routed_values = self.resolve_host_variable(var_name)
+        if value:
+            return [host for host in hosts if routed_values.get(host.name) == value]
+        return [host for host in hosts if host.name not in routed_values]
+
+    def get_script_data(self, hostvars=False, towervars=False, show_all=False, slice_number=1, slice_count=1, ig_routing_var=None, ig_routing_value=None):
         hosts_kw = dict()
         if not show_all:
             hosts_kw['enabled'] = True
@@ -309,6 +403,9 @@ def get_script_data(self, hostvars=False, towervars=False, show_all=False, slice
             fetch_fields.append('enabled')
         host_queryset = self.hosts.filter(**hosts_kw).order_by('name').only(*fetch_fields)
         hosts = self.get_sliced_hosts(host_queryset, slice_number, slice_count)
+        if ig_routing_var:
+            # Restrict the inventory to the routing bucket of this job
+            hosts = self.filter_hosts_to_routing_bucket(hosts, ig_routing_var, ig_routing_value)
 
         data = dict()
         all_group = data.setdefault('all', dict())