Route hosts to instance groups from an inventory variable

awx/api/serializers.py

@@ -3201,6 +3201,7 @@ class Meta:
             'forks',
             'limit',
             'job_slice_pinned_hosts',
+            'instance_group_routing_var',
             'verbosity',
             'extra_vars',
             'job_tags',
@@ -3417,6 +3418,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):
@@ -3467,6 +3475,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

@@ -2436,6 +2436,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)
@@ -2857,6 +2865,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)
 
@@ -3269,6 +3278,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 jt.get_effective_slice_ct({'inventory': obj.inventory}) != obj.workflow_nodes.count()
             ):

awx/api/views/webhooks.py

@@ -4,6 +4,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
@@ -181,7 +182,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/0205_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', '0204_job_slice_pinned_hosts'),
+    ]
+
+    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

@@ -309,7 +309,111 @@ def get_sliced_hosts(self, host_queryset, slice_number, slice_count, pinned_host
             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, slice_pinned_hosts=None):
+    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,
+        slice_pinned_hosts=None,
+        ig_routing_var=None,
+        ig_routing_value=None,
+    ):
         hosts_kw = dict()
         if not show_all:
             hosts_kw['enabled'] = True
@@ -318,6 +422,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, pinned_hosts=slice_pinned_hosts)
+        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())