diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index 76ca9c17..daee839d 100644
--- a/awx/api/serializers.py
+++ b/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',
diff --git a/awx/api/views/__init__.py b/awx/api/views/__init__.py
index 42a4151f..71e6afe2 100644
--- a/awx/api/views/__init__.py
+++ b/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()
):
diff --git a/awx/api/views/webhooks.py b/awx/api/views/webhooks.py
index 8af2d122..59c1e7d2 100644
--- a/awx/api/views/webhooks.py
+++ b/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)
diff --git a/awx/main/migrations/0205_instance_group_routing.py b/awx/main/migrations/0205_instance_group_routing.py
new file mode 100644
index 00000000..7fb18d65
--- /dev/null
+++ b/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,
+ ),
+ ),
+ ]
diff --git a/awx/main/models/inventory.py b/awx/main/models/inventory.py
index 7f07b6c3..c1992145 100644
--- a/awx/main/models/inventory.py
+++ b/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())
diff --git a/awx/main/models/jobs.py b/awx/main/models/jobs.py
index cb821fb5..4df20f4c 100644
--- a/awx/main/models/jobs.py
+++ b/awx/main/models/jobs.py
@@ -121,6 +121,20 @@ class Meta:
'not supported. Has no effect unless the job is sliced.'
),
)
+ instance_group_routing_var = models.CharField(
+ max_length=1024,
+ 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.'
+ ),
+ )
verbosity = models.PositiveIntegerField(
choices=VERBOSITY_CHOICES,
blank=True,
@@ -381,10 +395,13 @@ def create_job(self, **kwargs):
"""
return self.create_unified_job(**kwargs)
- def get_effective_slice_ct(self, kwargs):
- actual_inventory = self.inventory
+ def _get_actual_inventory(self, kwargs):
if self.ask_inventory_on_launch and 'inventory' in kwargs:
- actual_inventory = kwargs['inventory']
+ return kwargs['inventory']
+ return self.inventory
+
+ def get_effective_slice_ct(self, kwargs):
+ actual_inventory = self._get_actual_inventory(kwargs)
actual_slice_count = self.job_slice_count
if self.ask_job_slice_count_on_launch and 'job_slice_count' in kwargs:
actual_slice_count = kwargs['job_slice_count']
@@ -400,6 +417,72 @@ def get_effective_slice_ct(self, kwargs):
else:
return actual_slice_count
+ def get_ig_routing_buckets(self, inventory):
+ """
+ Group the inventory hosts by the value of the routing variable.
+ Returns a list of (value, InstanceGroup) tuples, one per distinct value,
+ plus a final ('', None) bucket if any host does not resolve the variable.
+ Returns an empty list when no host resolves it, so routing is a no-op.
+ Raises ValidationError if a value does not name an existing instance group.
+ """
+ from awx.main.models.ha import InstanceGroup
+
+ routed_values = inventory.resolve_host_variable(self.instance_group_routing_var)
+ values = sorted(set(routed_values.values()))
+ if not values:
+ return []
+ instance_groups = {ig.name: ig for ig in InstanceGroup.objects.filter(name__in=values)}
+ missing = [value for value in values if value not in instance_groups]
+ if missing:
+ raise ValidationError(
+ _('Instance group routing variable "%(var)s" resolves to instance groups that do not exist: %(names)s')
+ % {'var': self.instance_group_routing_var, 'names': ', '.join(missing)}
+ )
+ buckets = [(value, instance_groups[value]) for value in values]
+ if len(routed_values) < inventory.hosts.filter(enabled=True).count():
+ buckets.append(('', None))
+ return buckets
+
+ def get_ig_routing_launch_error(self, user, kwargs):
+ """
+ Validate that launching with the given prompts may route jobs to the
+ instance groups the inventory names. Returns (error, buckets): an error
+ message (or None) and the computed routing buckets (or None when routing
+ does not apply), so the caller can hand the validated buckets back to
+ create_unified_job and avoid recomputing them. The use_role check
+ mirrors the one JobLaunchConfigAccess applies to instance groups
+ prompted at launch.
+ """
+ from awx.main.models.ha import InstanceGroup
+
+ if not self.instance_group_routing_var:
+ return None, None
+ if kwargs.get('instance_groups'):
+ return None, None
+ if self.get_effective_slice_ct(kwargs) > 1:
+ return _('Instance group routing cannot be combined with job slicing. Launch with a job slice count of 1.'), None
+ actual_inventory = self._get_actual_inventory(kwargs)
+ if actual_inventory is None or getattr(actual_inventory, 'kind', None) == 'federated':
+ return None, None
+ try:
+ buckets = self.get_ig_routing_buckets(actual_inventory)
+ except ValidationError as exc:
+ return ' '.join(exc.messages), None
+ if user is None or user.is_superuser:
+ return None, buckets
+ routed_igs = [instance_group for value, instance_group in buckets if instance_group is not None]
+ if not routed_igs:
+ return None, buckets
+ accessible_qs = InstanceGroup.accessible_pk_qs(user, 'use_role')
+ denied = sorted(
+ InstanceGroup.objects.filter(pk__in=[instance_group.pk for instance_group in routed_igs])
+ .exclude(pk__in=accessible_qs)
+ .values_list('name', flat=True)
+ )
+ if denied:
+ return _('You do not have use permission on instance groups this job would be routed to: %(names)s') % {'names': ', '.join(denied)}, None
+ return None, buckets
+
def save(self, *args, **kwargs):
update_fields = kwargs.get('update_fields', [])
# if project is deleted for some reason, then keep the old organization
@@ -433,32 +516,51 @@ def validate_unique(self, exclude=None):
def create_unified_job(self, **kwargs):
prevent_slicing = kwargs.pop('_prevent_slicing', False)
prevent_federation = kwargs.pop('_prevent_federation', False)
+ prevent_ig_routing = kwargs.pop('_prevent_ig_routing', False)
+ # buckets already computed (and RBAC checked) by the launch endpoint
+ precomputed_routing_buckets = kwargs.pop('_ig_routing_buckets', None)
- # Determine the effective inventory for federation detection
- actual_inventory = self.inventory
- if self.ask_inventory_on_launch and 'inventory' in kwargs:
- actual_inventory = kwargs['inventory']
+ # Determine the effective inventory for federation and routing detection
+ actual_inventory = self._get_actual_inventory(kwargs)
federated_event = bool(actual_inventory and getattr(actual_inventory, 'kind', None) == 'federated' and not prevent_federation)
slice_ct = self.get_effective_slice_ct(kwargs)
slice_event = bool(slice_ct > 1 and (not prevent_slicing))
- if federated_event:
- # A Federated Inventory generates a WorkflowJob with one node per source inventory,
- # each inheriting that inventory's instance groups for automatic IG routing.
- from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobNode
- kwargs['_unified_job_class'] = WorkflowJobTemplate._get_unified_job_class()
- kwargs['_parent_field_name'] = "job_template"
- kwargs.setdefault('_eager_fields', {})
- kwargs['_eager_fields']['is_sliced_job'] = True
- elif slice_event:
- # A Slice Job Template will generate a WorkflowJob rather than a Job
+ routing_buckets = []
+ if (
+ self.instance_group_routing_var
+ and not prevent_ig_routing
+ and not federated_event
+ and not slice_event
+ # instance groups given at launch are an explicit override of the routing
+ and not kwargs.get('instance_groups')
+ and actual_inventory is not None
+ and getattr(actual_inventory, 'kind', None) != 'federated'
+ ):
+ if precomputed_routing_buckets is not None:
+ routing_buckets = precomputed_routing_buckets
+ else:
+ routing_buckets = self.get_ig_routing_buckets(actual_inventory)
+ routing_event = bool(len(routing_buckets) > 1)
+
+ if federated_event or slice_event or routing_event:
+ # Federated inventories, sliced jobs and instance group routing all
+ # generate a WorkflowJob rather than a Job: one node per source
+ # inventory, per slice, or per routing bucket respectively.
from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobNode
kwargs['_unified_job_class'] = WorkflowJobTemplate._get_unified_job_class()
kwargs['_parent_field_name'] = "job_template"
kwargs.setdefault('_eager_fields', {})
kwargs['_eager_fields']['is_sliced_job'] = True
+ elif len(routing_buckets) == 1:
+ # All hosts routed to the same place, launch a plain job on it
+ value, instance_group = routing_buckets[0]
+ kwargs.setdefault('_eager_fields', {})
+ kwargs['_eager_fields']['instance_group_routing_value'] = value
+ if instance_group is not None:
+ kwargs['instance_groups'] = [instance_group]
elif self.job_slice_count > 1 and (not prevent_slicing):
# Unique case where JT was set to slice but hosts not available
kwargs.setdefault('_eager_fields', {})
@@ -483,6 +585,12 @@ def create_unified_job(self, **kwargs):
for idx in range(slice_ct):
create_kwargs = dict(workflow_job=job, unified_job_template=self, ancestor_artifacts=dict(job_slice=idx + 1))
WorkflowJobNode.objects.create(**create_kwargs)
+ elif routing_event:
+ for value, instance_group in routing_buckets:
+ ancestor_artifacts = dict(ig_routing_value=value)
+ if instance_group is not None:
+ ancestor_artifacts['ig_routing_instance_group_id'] = instance_group.id
+ WorkflowJobNode.objects.create(workflow_job=job, unified_job_template=self, ancestor_artifacts=ancestor_artifacts)
return job
def get_absolute_url(self, request=None):
@@ -691,6 +799,15 @@ class Meta:
default=1,
help_text=_("If ran as part of sliced jobs, the total number of slices. If 1, job is not part of a sliced job."),
)
+ instance_group_routing_value = models.TextField(
+ null=True,
+ 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."
+ ),
+ )
def _get_parent_field_name(self):
return 'job_template'
@@ -738,12 +855,26 @@ def copy_unified_job(self, **new_prompts):
# target same slice as original job
new_prompts['_prevent_slicing'] = True
new_prompts['_prevent_federation'] = True
+ new_prompts['_prevent_ig_routing'] = True
new_prompts.setdefault('_eager_fields', {})
new_prompts['_eager_fields']['inventory_id'] = self.inventory_id
new_prompts['_eager_fields']['job_slice_number'] = self.job_slice_number
new_prompts['_eager_fields']['job_slice_count'] = self.job_slice_count
+ # target the same routing bucket as the original job, even if the routing
+ # var changed on the template since; the routed instance group comes back
+ # through the launch config prompts
+ new_prompts['_eager_fields']['instance_group_routing_var'] = self.instance_group_routing_var
+ new_prompts['_eager_fields']['instance_group_routing_value'] = self.instance_group_routing_value
return super(Job, self).copy_unified_job(**new_prompts)
+ @property
+ def is_ig_routed(self):
+ return bool(self.instance_group_routing_var) and self.instance_group_routing_value is not None
+
+ def filter_hosts_to_ig_routing_bucket(self, hosts):
+ """Restrict hosts to the ones belonging to this routed job's bucket."""
+ return self.inventory.filter_hosts_to_routing_bucket(hosts, self.instance_group_routing_var, self.instance_group_routing_value)
+
def get_passwords_needed_to_start(self):
return self.passwords_needed_to_start
@@ -787,6 +918,8 @@ def _get_task_impact(self):
count_hosts = (count_hosts + self.job_slice_count - self.job_slice_number) // self.job_slice_count
# pinned hosts run in every slice on top of its share
count_hosts += pinned_ct
+ elif self.is_ig_routed:
+ count_hosts = len(self.filter_hosts_to_ig_routing_bucket(self.inventory.hosts.filter(enabled=True).only('id', 'name')))
else:
count_hosts = 5 if self.forks == 0 else self.forks
return min(count_hosts, 5 if self.forks == 0 else self.forks) + 1
@@ -949,6 +1082,9 @@ def get_hosts_for_fact_cache(self):
host_qs = host_qs.only(*HOST_FACTS_FIELDS)
host_qs = self.inventory.get_sliced_hosts(host_qs, self.job_slice_number, self.job_slice_count, pinned_hosts=self.job_slice_pinned_hosts_list)
+ if self.is_ig_routed:
+ # routed jobs only run their bucket of hosts, keep the fact cache aligned
+ host_qs = self.filter_hosts_to_ig_routing_bucket(host_qs)
return host_qs
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index fce8fdd2..b1ab7059 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -625,6 +625,22 @@ def get_job_kwargs(self):
data['_eager_fields']['allow_simultaneous'] = True
data['_eager_fields']['inventory_id'] = self.ancestor_artifacts['source_inventory_id']
data['_prevent_federation'] = True
+ # Extra processing for instance group routing: each node runs the bucket of
+ # hosts that resolved to its routing value, on the routed instance group.
+ if 'ig_routing_value' in self.ancestor_artifacts and is_root_node:
+ data['_eager_fields']['allow_simultaneous'] = True
+ data['_eager_fields']['instance_group_routing_value'] = self.ancestor_artifacts['ig_routing_value']
+ data['_prevent_slicing'] = True
+ data['_prevent_ig_routing'] = True
+ routed_ig_id = self.ancestor_artifacts.get('ig_routing_instance_group_id')
+ if routed_ig_id:
+ from awx.main.models.ha import InstanceGroup
+
+ routed_ig = InstanceGroup.objects.filter(pk=routed_ig_id).first()
+ if routed_ig is not None:
+ data['instance_groups'] = [routed_ig]
+ else:
+ logger.warning('Instance group %s routed at launch no longer exists, job falls back to default instance groups.', routed_ig_id)
return data
diff --git a/awx/main/scheduler/task_manager.py b/awx/main/scheduler/task_manager.py
index 91859672..20bce47d 100644
--- a/awx/main/scheduler/task_manager.py
+++ b/awx/main/scheduler/task_manager.py
@@ -11,6 +11,7 @@
import signal
# Django
+from django.core.exceptions import ValidationError as DjangoValidationError
from django.db import transaction
from django.utils.translation import gettext_lazy as _, gettext_noop
from django.utils.timezone import now as tz_now
@@ -229,7 +230,23 @@ def spawn_workflow_graph_jobs(self):
superseded_job_id = spawn_node.job_id
is_retry = superseded_job_id is not None
kv = spawn_node.get_job_kwargs()
- job = spawn_node.unified_job_template.create_unified_job(**kv)
+ try:
+ job = spawn_node.unified_job_template.create_unified_job(**kv)
+ except DjangoValidationError as exc:
+ # e.g. instance group routing pointing at an instance group
+ # that does not exist; fail this workflow loudly instead of
+ # crashing the scheduling cycle for every workflow
+ logger.warning('Failed to spawn job in %s for node %s: %s', workflow_job.log_format, spawn_node.pk, ' '.join(exc.messages))
+ workflow_job.status = 'failed'
+ workflow_job.job_explanation = gettext_noop("Workflow job could not spawn a job for one of its nodes: {}").format(
+ ' '.join(exc.messages)
+ )
+ workflow_job.start_args = '' # blank field to remove encrypted passwords
+ workflow_job.save(update_fields=['status', 'job_explanation', 'start_args'])
+ workflow_job.websocket_emit_status('failed')
+ workflow_job.send_notification_templates('failed')
+ result.append(workflow_job.id)
+ break
spawn_node.job = job
if is_retry:
spawn_node.retry_attempts += 1
diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py
index 86ef2a7f..bbd885d3 100644
--- a/awx/main/tasks/jobs.py
+++ b/awx/main/tasks/jobs.py
@@ -328,6 +328,9 @@ def build_inventory(self, instance, private_data_dir):
script_params['slice_number'] = instance.job_slice_number
script_params['slice_count'] = instance.job_slice_count
script_params['slice_pinned_hosts'] = instance.job_slice_pinned_hosts_list
+ if getattr(instance, 'is_ig_routed', False):
+ script_params['ig_routing_var'] = instance.instance_group_routing_var
+ script_params['ig_routing_value'] = instance.instance_group_routing_value
return self.write_inventory_file(instance.inventory, private_data_dir, 'hosts', script_params)
diff --git a/awx/main/tests/functional/api/test_instance_group_routing.py b/awx/main/tests/functional/api/test_instance_group_routing.py
new file mode 100644
index 00000000..dabad65b
--- /dev/null
+++ b/awx/main/tests/functional/api/test_instance_group_routing.py
@@ -0,0 +1,108 @@
+import pytest
+
+from awx.api.versioning import reverse
+from awx.main.models import JobTemplate, InstanceGroup, WorkflowJob
+
+
+@pytest.fixture
+def routed_inventory(inventory):
+ dc1 = inventory.groups.create(name='datacenter1', variables={'dc_instance_group': 'dc1-nodes'})
+ dc2 = inventory.groups.create(name='datacenter2', variables={'dc_instance_group': 'dc2-nodes'})
+ dc1.hosts.add(inventory.hosts.create(name='dc1-host'))
+ dc2.hosts.add(inventory.hosts.create(name='dc2-host'))
+ inventory.hosts.create(name='lonely-host')
+ return inventory
+
+
+@pytest.fixture
+def routed_igs():
+ return {name: InstanceGroup.objects.create(name=name) for name in ('dc1-nodes', 'dc2-nodes')}
+
+
+@pytest.fixture
+def routed_jt(project, routed_inventory):
+ return JobTemplate.objects.create(
+ name='routed-jt',
+ project=project,
+ inventory=routed_inventory,
+ playbook='helloworld.yml',
+ instance_group_routing_var='dc_instance_group',
+ )
+
+
+@pytest.mark.django_db
+class TestInstanceGroupRoutingLaunch:
+ def test_launch_denied_without_use_role(self, routed_jt, routed_igs, rando, post):
+ routed_jt.execute_role.members.add(rando)
+ response = post(reverse('api:job_template_launch', kwargs={'pk': routed_jt.pk}), {}, rando, expect=400)
+ assert 'use permission' in response.data['errors'][0]
+ assert 'dc1-nodes' in response.data['errors'][0]
+
+ def test_launch_allowed_with_use_role(self, routed_jt, routed_igs, rando, post, mocker):
+ routed_jt.execute_role.members.add(rando)
+ for instance_group in routed_igs.values():
+ instance_group.use_role.members.add(rando)
+ mocker.patch.object(WorkflowJob, 'signal_start', return_value=True)
+ response = post(reverse('api:job_template_launch', kwargs={'pk': routed_jt.pk}), {}, rando, expect=201)
+ assert 'workflow_job' in response.data
+ workflow_job = WorkflowJob.objects.get(pk=response.data['workflow_job'])
+ assert workflow_job.workflow_nodes.count() == 3
+
+ def test_launch_allowed_for_superuser(self, routed_jt, routed_igs, admin_user, post, mocker):
+ mocker.patch.object(WorkflowJob, 'signal_start', return_value=True)
+ response = post(reverse('api:job_template_launch', kwargs={'pk': routed_jt.pk}), {}, admin_user, expect=201)
+ assert 'workflow_job' in response.data
+
+ def test_launch_rejected_when_slicing_prompted(self, routed_jt, routed_igs, admin_user, post):
+ routed_jt.ask_job_slice_count_on_launch = True
+ routed_jt.save()
+ response = post(reverse('api:job_template_launch', kwargs={'pk': routed_jt.pk}), {'job_slice_count': 3}, admin_user, expect=400)
+ assert 'job slicing' in str(response.data)
+
+ def test_launch_rejected_on_unknown_instance_group(self, routed_jt, routed_igs, admin_user, post):
+ routed_igs['dc2-nodes'].delete()
+ response = post(reverse('api:job_template_launch', kwargs={'pk': routed_jt.pk}), {}, admin_user, expect=400)
+ assert 'dc2-nodes' in response.data['errors'][0]
+
+
+@pytest.mark.django_db
+class TestInstanceGroupRoutingValidation:
+ def test_routing_var_must_be_a_variable_name(self, routed_jt, admin_user, patch):
+ url = reverse('api:job_template_detail', kwargs={'pk': routed_jt.pk})
+ response = patch(url, {'instance_group_routing_var': 'not a var!'}, admin_user, expect=400)
+ assert 'valid variable name' in str(response.data['instance_group_routing_var'])
+
+ def test_routing_cannot_be_combined_with_slicing(self, routed_jt, admin_user, patch):
+ url = reverse('api:job_template_detail', kwargs={'pk': routed_jt.pk})
+ response = patch(url, {'job_slice_count': 3}, admin_user, expect=400)
+ assert 'job slicing' in str(response.data['instance_group_routing_var'])
+
+ def test_routing_var_accepted(self, project, inventory, admin_user, patch):
+ jt = JobTemplate.objects.create(name='plain-jt', project=project, inventory=inventory, playbook='helloworld.yml')
+ url = reverse('api:job_template_detail', kwargs={'pk': jt.pk})
+ patch(url, {'instance_group_routing_var': 'dc_instance_group'}, admin_user, expect=200)
+
+
+@pytest.mark.django_db
+class TestInstanceGroupRoutingWorkflowRelaunch:
+ def test_relaunch_recomputes_buckets(self, routed_jt, routed_igs, admin_user, post, mocker):
+ mocker.patch.object(WorkflowJob, 'signal_start', return_value=True)
+ workflow_job = routed_jt.create_unified_job()
+ url = reverse('api:workflow_job_relaunch', kwargs={'pk': workflow_job.pk})
+ response = post(url, {}, admin_user, expect=201)
+ relaunched = WorkflowJob.objects.get(pk=response.data['id'])
+ assert relaunched.workflow_nodes.count() == 3
+
+ def test_relaunch_rejected_when_routing_collapses(self, routed_jt, routed_igs, admin_user, post):
+ workflow_job = routed_jt.create_unified_job()
+ routed_jt.inventory.hosts.exclude(name='dc1-host').delete()
+ url = reverse('api:workflow_job_relaunch', kwargs={'pk': workflow_job.pk})
+ response = post(url, {}, admin_user, expect=400)
+ assert 'no longer routes' in str(response.data)
+
+ def test_relaunch_rejected_without_use_role(self, routed_jt, routed_igs, rando, admin_user, post):
+ workflow_job = routed_jt.create_unified_job()
+ routed_jt.execute_role.members.add(rando)
+ url = reverse('api:workflow_job_relaunch', kwargs={'pk': workflow_job.pk})
+ response = post(url, {}, rando, expect=400)
+ assert 'use permission' in str(response.data)
diff --git a/awx/main/tests/functional/models/test_inventory.py b/awx/main/tests/functional/models/test_inventory.py
index 8728925c..baed9e32 100644
--- a/awx/main/tests/functional/models/test_inventory.py
+++ b/awx/main/tests/functional/models/test_inventory.py
@@ -334,3 +334,80 @@ def test_host_distinctness(self, setup_inventory_groups, organization):
# 2 organizations with host of same name only has 1 entry in smart inventory
# smart inventory in 1 organization does not include host from another
# smart inventory correctly returns hosts in filter in same organization
+
+
+@pytest.mark.django_db
+class TestResolveHostVariable:
+ def test_host_variable_wins_over_group(self, inventory):
+ group = inventory.groups.create(name='dc1', variables={'route_var': 'group-value'})
+ host = inventory.hosts.create(name='host1', variables={'route_var': 'host-value'})
+ group.hosts.add(host)
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'host-value'}
+
+ def test_group_variable_reaches_hosts(self, inventory):
+ group = inventory.groups.create(name='dc1', variables={'route_var': 'dc1-nodes'})
+ group.hosts.add(inventory.hosts.create(name='host1'))
+ group.hosts.add(inventory.hosts.create(name='host2'))
+ inventory.hosts.create(name='ungrouped-host')
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'dc1-nodes', 'host2': 'dc1-nodes'}
+
+ def test_variable_inherited_from_parent_group(self, inventory):
+ parent = inventory.groups.create(name='region', variables={'route_var': 'region-nodes'})
+ child = inventory.groups.create(name='rack')
+ parent.children.add(child)
+ child.hosts.add(inventory.hosts.create(name='host1'))
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'region-nodes'}
+
+ def test_deeper_group_overrides_parent(self, inventory):
+ parent = inventory.groups.create(name='region', variables={'route_var': 'region-nodes'})
+ child = inventory.groups.create(name='aisle', variables={'route_var': 'aisle-nodes'})
+ parent.children.add(child)
+ child.hosts.add(inventory.hosts.create(name='host1'))
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'aisle-nodes'}
+
+ def test_ansible_group_priority_wins_over_name_order(self, inventory):
+ g1 = inventory.groups.create(name='alpha', variables={'route_var': 'alpha-nodes', 'ansible_group_priority': 10})
+ g2 = inventory.groups.create(name='beta', variables={'route_var': 'beta-nodes'})
+ host = inventory.hosts.create(name='host1')
+ g1.hosts.add(host)
+ g2.hosts.add(host)
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'alpha-nodes'}
+
+ def test_disabled_hosts_are_ignored(self, inventory):
+ group = inventory.groups.create(name='dc1', variables={'route_var': 'dc1-nodes'})
+ group.hosts.add(inventory.hosts.create(name='on-host'))
+ group.hosts.add(inventory.hosts.create(name='off-host', enabled=False))
+ assert inventory.resolve_host_variable('route_var') == {'on-host': 'dc1-nodes'}
+
+ def test_same_depth_groups_merge_in_name_order(self, inventory):
+ g1 = inventory.groups.create(name='alpha', variables={'route_var': 'alpha-nodes'})
+ g2 = inventory.groups.create(name='beta', variables={'route_var': 'beta-nodes'})
+ host = inventory.hosts.create(name='host1')
+ g1.hosts.add(host)
+ g2.hosts.add(host)
+ assert inventory.resolve_host_variable('route_var') == {'host1': 'beta-nodes'}
+
+ def test_non_string_and_empty_values_are_ignored(self, inventory):
+ inventory.hosts.create(name='dict-host', variables={'route_var': {'nested': 'value'}})
+ inventory.hosts.create(name='int-host', variables={'route_var': 5})
+ inventory.hosts.create(name='empty-host', variables={'route_var': ''})
+ inventory.hosts.create(name='plain-host')
+ assert inventory.resolve_host_variable('route_var') == {}
+
+ def test_invalid_host_value_does_not_fall_back_to_group(self, inventory):
+ group = inventory.groups.create(name='dc1', variables={'route_var': 'dc1-nodes'})
+ host = inventory.hosts.create(name='host1', variables={'route_var': ['not', 'a', 'string']})
+ group.hosts.add(host)
+ assert inventory.resolve_host_variable('route_var') == {}
+
+ def test_script_data_filters_to_routing_bucket(self, inventory):
+ group = inventory.groups.create(name='dc1', variables={'route_var': 'dc1-nodes'})
+ group.hosts.add(inventory.hosts.create(name='routed-host'))
+ inventory.hosts.create(name='fallback-host')
+ routed = inventory.get_script_data(hostvars=True, ig_routing_var='route_var', ig_routing_value='dc1-nodes')
+ assert set(routed['_meta']['hostvars'].keys()) == {'routed-host'}
+ assert routed['dc1']['hosts'] == ['routed-host']
+ fallback = inventory.get_script_data(hostvars=True, ig_routing_var='route_var', ig_routing_value='')
+ assert set(fallback['_meta']['hostvars'].keys()) == {'fallback-host'}
+ # the group stays (it still carries vars) but its routed host is gone
+ assert 'hosts' not in fallback.get('dc1', {})
diff --git a/awx/main/tests/functional/models/test_job.py b/awx/main/tests/functional/models/test_job.py
index 704d549e..e9faa7ea 100644
--- a/awx/main/tests/functional/models/test_job.py
+++ b/awx/main/tests/functional/models/test_job.py
@@ -1,6 +1,8 @@
import pytest
-from awx.main.models import JobTemplate, Job, JobHostSummary, WorkflowJob, Inventory, Project, Organization
+from django.core.exceptions import ValidationError
+
+from awx.main.models import JobTemplate, Job, JobHostSummary, WorkflowJob, Inventory, Project, Organization, InstanceGroup
from awx.main.models.jobs import _federated_inventory_has_matching_hosts
@@ -359,3 +361,158 @@ def test_complex_pattern_failsafe(self, organization):
assert _federated_inventory_has_matching_hosts(inv, 'web:db') is True
assert _federated_inventory_has_matching_hosts(inv, 'web&db') is True
assert _federated_inventory_has_matching_hosts(inv, '!web') is True
+
+
+@pytest.mark.django_db
+class TestInstanceGroupRouting:
+ @pytest.fixture
+ def routed_inventory(self, inventory):
+ dc1 = inventory.groups.create(name='datacenter1', variables={'dc_instance_group': 'dc1-nodes'})
+ dc2 = inventory.groups.create(name='datacenter2', variables={'dc_instance_group': 'dc2-nodes'})
+ for i in range(2):
+ dc1.hosts.add(inventory.hosts.create(name='dc1-host{}'.format(i)))
+ dc2.hosts.add(inventory.hosts.create(name='dc2-host{}'.format(i)))
+ inventory.hosts.create(name='lonely-host')
+ return inventory
+
+ @pytest.fixture
+ def routed_igs(self):
+ return {name: InstanceGroup.objects.create(name=name) for name in ('dc1-nodes', 'dc2-nodes')}
+
+ @pytest.fixture
+ def routed_jt(self, routed_inventory):
+ return JobTemplate.objects.create(name='routed-jt', inventory=routed_inventory, instance_group_routing_var='dc_instance_group')
+
+ @staticmethod
+ def spawn_nodes(workflow_job):
+ jobs = []
+ for node in workflow_job.workflow_nodes.all():
+ # does what the task manager does for spawning workflow jobs
+ kv = node.get_job_kwargs()
+ job = node.unified_job_template.create_unified_job(**kv)
+ node.job = job
+ node.save()
+ jobs.append(job)
+ return jobs
+
+ def test_routing_creates_workflow_with_buckets(self, routed_jt, routed_igs):
+ workflow_job = routed_jt.create_unified_job()
+ assert isinstance(workflow_job, WorkflowJob)
+ assert workflow_job.is_sliced_job
+ artifacts = sorted((node.ancestor_artifacts for node in workflow_job.workflow_nodes.all()), key=lambda a: a['ig_routing_value'])
+ assert artifacts == [
+ {'ig_routing_value': ''},
+ {'ig_routing_value': 'dc1-nodes', 'ig_routing_instance_group_id': routed_igs['dc1-nodes'].id},
+ {'ig_routing_value': 'dc2-nodes', 'ig_routing_instance_group_id': routed_igs['dc2-nodes'].id},
+ ]
+
+ def test_routed_jobs_get_bucket_and_instance_group(self, routed_jt, routed_igs):
+ workflow_job = routed_jt.create_unified_job()
+ jobs = {job.instance_group_routing_value: job for job in self.spawn_nodes(workflow_job)}
+ assert set(jobs) == {'', 'dc1-nodes', 'dc2-nodes'}
+ for value, job in jobs.items():
+ assert job.instance_group_routing_var == 'dc_instance_group'
+ assert job.allow_simultaneous
+ script_data = job.inventory.get_script_data(hostvars=True, ig_routing_var=job.instance_group_routing_var, ig_routing_value=value)
+ hostnames = set(script_data['_meta']['hostvars'].keys())
+ if value:
+ prefix = value.split('-')[0]
+ assert hostnames == {'{}-host0'.format(prefix), '{}-host1'.format(prefix)}
+ assert job.preferred_instance_groups_cache == [routed_igs[value].id]
+ else:
+ assert hostnames == {'lonely-host'}
+ # the fallback bucket keeps the normal instance group selection
+ assert job.preferred_instance_groups_cache == job._get_preferred_instance_group_cache()
+
+ def test_single_bucket_launches_plain_job(self, inventory, routed_igs):
+ group = inventory.groups.create(name='datacenter1', variables={'dc_instance_group': 'dc1-nodes'})
+ group.hosts.add(inventory.hosts.create(name='only-host'))
+ jt = JobTemplate.objects.create(name='routed-jt', inventory=inventory, instance_group_routing_var='dc_instance_group')
+ job = jt.create_unified_job()
+ assert isinstance(job, Job)
+ assert job.instance_group_routing_value == 'dc1-nodes'
+ assert job.preferred_instance_groups_cache == [routed_igs['dc1-nodes'].id]
+
+ def test_no_resolving_hosts_is_a_noop(self, inventory):
+ inventory.hosts.create(name='plain-host')
+ jt = JobTemplate.objects.create(name='routed-jt', inventory=inventory, instance_group_routing_var='dc_instance_group')
+ job = jt.create_unified_job()
+ assert isinstance(job, Job)
+ assert job.instance_group_routing_value is None
+
+ def test_unknown_instance_group_raises(self, routed_jt, routed_igs):
+ routed_igs['dc2-nodes'].delete()
+ with pytest.raises(ValidationError) as excinfo:
+ routed_jt.create_unified_job()
+ assert 'dc2-nodes' in str(excinfo.value)
+
+ def test_prompted_instance_groups_skip_routing(self, routed_jt, routed_igs):
+ other_ig = InstanceGroup.objects.create(name='hand-picked')
+ routed_jt.ask_instance_groups_on_launch = True
+ routed_jt.save()
+ job = routed_jt.create_unified_job(instance_groups=[other_ig])
+ assert isinstance(job, Job)
+ assert job.instance_group_routing_value is None
+ assert job.preferred_instance_groups_cache == [other_ig.id]
+
+ def test_routed_job_task_impact(self, routed_jt, routed_igs):
+ routed_jt.inventory.update_computed_fields()
+ workflow_job = routed_jt.create_unified_job()
+ impacts = {job.instance_group_routing_value: job.task_impact for job in self.spawn_nodes(workflow_job)}
+ # two hosts per datacenter bucket, one host in the fallback bucket, plus one
+ assert impacts == {'dc1-nodes': 3, 'dc2-nodes': 3, '': 2}
+
+ def test_routed_job_fact_cache_alignment(self, routed_jt, routed_igs):
+ routed_jt.use_fact_cache = True
+ routed_jt.save()
+ workflow_job = routed_jt.create_unified_job()
+ for job in self.spawn_nodes(workflow_job):
+ script_data = job.inventory.get_script_data(
+ hostvars=True, ig_routing_var=job.instance_group_routing_var, ig_routing_value=job.instance_group_routing_value
+ )
+ script_hosts = set(script_data['_meta']['hostvars'].keys())
+ fact_hosts = set(host.name for host in job.get_hosts_for_fact_cache())
+ assert fact_hosts == script_hosts
+
+ def test_relaunch_of_routed_child_keeps_bucket(self, routed_jt, routed_igs):
+ workflow_job = routed_jt.create_unified_job()
+ original = {job.instance_group_routing_value: job for job in self.spawn_nodes(workflow_job)}['dc1-nodes']
+ # clearing the routing var on the template must not widen the relaunch
+ # to the whole inventory while it stays pinned to the bucket's group
+ routed_jt.instance_group_routing_var = ''
+ routed_jt.save()
+ relaunched = original.copy_unified_job()
+ assert isinstance(relaunched, Job)
+ assert relaunched.instance_group_routing_var == 'dc_instance_group'
+ assert relaunched.instance_group_routing_value == 'dc1-nodes'
+ assert relaunched.is_ig_routed
+ assert relaunched.preferred_instance_groups_cache == [routed_igs['dc1-nodes'].id]
+
+ def test_prevent_ig_routing_launches_plain_job(self, routed_jt, routed_igs):
+ job = routed_jt.create_unified_job(_prevent_ig_routing=True)
+ assert isinstance(job, Job)
+ assert job.instance_group_routing_value is None
+
+ def test_disabled_hosts_do_not_create_buckets(self, routed_jt, routed_igs):
+ # a disabled host is the only one routing to a value whose instance
+ # group is gone: it must not break the launch nor create a bucket
+ stale = routed_jt.inventory.hosts.create(name='stale-host', enabled=False, variables={'dc_instance_group': 'gone-nodes'})
+ workflow_job = routed_jt.create_unified_job()
+ values = sorted(node.ancestor_artifacts['ig_routing_value'] for node in workflow_job.workflow_nodes.all())
+ assert values == ['', 'dc1-nodes', 'dc2-nodes']
+ stale.delete()
+
+ def test_precomputed_buckets_are_used(self, routed_jt, routed_igs):
+ buckets = routed_jt.get_ig_routing_buckets(routed_jt.inventory)
+ # drop the fallback bucket to prove the passed buckets win over recomputation
+ buckets = [(value, ig) for value, ig in buckets if value]
+ workflow_job = routed_jt.create_unified_job(_ig_routing_buckets=buckets)
+ assert workflow_job.workflow_nodes.count() == 2
+
+ def test_slicing_takes_precedence_over_routing(self, routed_jt, routed_igs):
+ routed_jt.job_slice_count = 2
+ routed_jt.save()
+ workflow_job = routed_jt.create_unified_job()
+ assert workflow_job.workflow_nodes.count() == 2
+ artifacts = [node.ancestor_artifacts for node in workflow_job.workflow_nodes.all()]
+ assert all('job_slice' in artifact for artifact in artifacts)
diff --git a/awx/main/tests/functional/task_management/test_scheduler.py b/awx/main/tests/functional/task_management/test_scheduler.py
index 5af505fd..57558c99 100644
--- a/awx/main/tests/functional/task_management/test_scheduler.py
+++ b/awx/main/tests/functional/task_management/test_scheduler.py
@@ -164,6 +164,27 @@ def test_workflow_node_never_started_job_is_not_retried(self, inventory, project
assert node.retry_attempts == node.max_retries
assert jt.jobs.count() == 1
+ def test_workflow_fails_when_routed_node_cannot_spawn(self, inventory, project, controlplane_instance_group):
+ # a routed JT inside a workflow whose routing value names a missing
+ # instance group must fail the workflow, not crash the manager cycle
+ group = inventory.groups.create(name='dc1', variables={'dc_instance_group': 'missing-nodes'})
+ group.hosts.add(inventory.hosts.create(name='host1'))
+ inventory.hosts.create(name='host2')
+ jt = JobTemplate.objects.create(
+ name='routed-in-wf', inventory=inventory, project=project, playbook='helloworld.yml', instance_group_routing_var='dc_instance_group'
+ )
+ wfjt = WorkflowJobTemplate.objects.create(name='wf-routed')
+ wfjt.workflow_nodes.create(unified_job_template=jt)
+ wj = wfjt.create_unified_job()
+ wj.signal_start()
+
+ self.run_tm(TaskManager(), [mock.call('running')])
+ self.run_tm(WorkflowManager(), [mock.call('failed')])
+
+ wj.refresh_from_db()
+ assert wj.status == 'failed'
+ assert 'missing-nodes' in wj.job_explanation
+
def test_task_manager_workflow_workflow_rescheduling(self, controlplane_instance_group):
wfjts = [WorkflowJobTemplate.objects.create(name='foo')]
for i in range(5):
diff --git a/awx/ui/src/screens/Template/JobTemplateDetail/JobTemplateDetail.js b/awx/ui/src/screens/Template/JobTemplateDetail/JobTemplateDetail.js
index 3869894c..c7d437ab 100644
--- a/awx/ui/src/screens/Template/JobTemplateDetail/JobTemplateDetail.js
+++ b/awx/ui/src/screens/Template/JobTemplateDetail/JobTemplateDetail.js
@@ -46,6 +46,7 @@ function JobTemplateDetail({ template }) {
extra_vars,
forks,
host_config_key,
+ instance_group_routing_var,
job_slice_count,
job_slice_pinned_hosts,
job_tags,
@@ -296,6 +297,12 @@ function JobTemplateDetail({ template }) {
dataCy="jt-detail-job-slice-pinned-hosts"
helpText={helpText.jobSlicePinnedHosts}
/>
+
{host_config_key && (
<>
)}
+
', () => {
expect(pinnedInput.value).toBe('localhost');
});
+ test('instance group routing variable field accepts a value', async () => {
+ renderWithContexts(
+
+ );
+ await screen.findByRole('button', { name: 'Save' });
+
+ const routingInput = document.getElementById(
+ 'template-instance-group-routing-var'
+ );
+ expect(routingInput).not.toBeNull();
+ fireEvent.change(routingInput, {
+ target: { value: 'dc_instance_group', name: 'instance_group_routing_var' },
+ });
+ expect(routingInput.value).toBe('dc_instance_group');
+ });
+
test('webhooks and enable concurrent jobs functions properly', async () => {
// WebhookSubForm reads the template id from useParams(), so mount the form
// under a real v6 route whose concrete URL provides id=1.
diff --git a/docs/instance_group_routing.md b/docs/instance_group_routing.md
new file mode 100644
index 00000000..b1435582
--- /dev/null
+++ b/docs/instance_group_routing.md
@@ -0,0 +1,89 @@
+# Instance Group Routing
+
+Instance group routing lets a single job template fan out over the instance groups
+that can actually reach each host, using data that already lives in the inventory.
+
+The typical setup: several isolated network zones (datacenters, security zones),
+each one with its own execution nodes collected in an instance group, because the
+firewalls only allow the execution nodes of a zone to reach the hosts of that zone.
+The playbooks are the same everywhere, but a job runs on a single instance group,
+so without routing every job template has to be duplicated per zone with a
+different `limit` and instance group.
+
+## How it works
+
+Set `instance_group_routing_var` on a job template to the name of an inventory
+variable, usually defined as a group variable:
+
+```yaml
+# group_vars for datacenter1
+dc_instance_group: dc1-nodes
+
+# group_vars for datacenter2
+dc_instance_group: dc2-nodes
+```
+
+At launch, hosts are grouped by the value that the variable resolves to for each
+of them. The launch then creates the same kind of implicit workflow job that
+sliced jobs create, with one node per distinct value:
+
+- Each node runs a job restricted to the hosts of its bucket (the rest of the
+ inventory is filtered out, the same way slicing filters each slice), assigned
+ to the instance group named by the value.
+- Hosts that do not resolve the variable run in one extra job that keeps the
+ normal instance group selection (job template, then inventory, then
+ organization, then the default).
+- If every host ends up in the same bucket, a plain job is launched instead of a
+ workflow, on the routed instance group.
+- If no host resolves the variable at all, the launch behaves as if routing were
+ not configured.
+
+Variable resolution follows a simplified version of the Ansible precedence
+rules: host variables win over group variables, and group values are merged
+sorted by depth, then `ansible_group_priority`, then group name, with later
+groups overriding earlier ones. Values must be non empty strings; anything
+else is treated as unset and the host goes to the fallback bucket. Disabled
+hosts are ignored, so they neither create buckets nor count for the fallback.
+
+## Security
+
+Instance groups are global objects protected by RBAC. The routing variable is
+inventory data, which can be edited by people with no permission at all on the
+instance groups, so it is not trusted blindly:
+
+- Routing only happens when the job template explicitly opts in by setting
+ `instance_group_routing_var`.
+- At launch, every instance group referenced by the inventory is checked against
+ the launching user's use permission, the same rule applied to instance groups
+ prompted at launch. If any referenced group fails the check, or does not
+ exist, the launch is rejected with an error naming the group. There is no
+ silent fallback.
+- Relaunching a routed workflow re-resolves the buckets and applies the same
+ checks against the relaunching user.
+
+Launches that do not go through the API launch endpoint (schedules, webhooks,
+workflow nodes) skip the per-user check, since there is no requesting user;
+they still fail loudly if a value does not name an existing instance group: a
+webhook launch answers with a 400, a workflow containing the routed template
+is marked failed with the reason in its explanation, and a scheduled launch
+is skipped with the error in the dispatcher log.
+
+## Interactions and limitations
+
+- Routing cannot be combined with job slicing: setting both
+ `instance_group_routing_var` and a `job_slice_count` greater than 1 is a
+ validation error, and prompting a slice count above 1 at launch time is
+ rejected by the launch endpoint. On launch paths without that check
+ (schedules, workflow nodes) slicing wins and routing is skipped.
+- Provisioning callbacks never route: a callback runs against the single
+ calling host, so the job uses the normal instance group selection, the same
+ way callbacks never slice.
+- Passing instance groups explicitly at launch (with
+ `ask_instance_groups_on_launch`) is treated as an override: the launch runs a
+ single job on the given instance groups and routing is skipped.
+- The buckets are computed from the whole inventory; a `limit` still applies
+ within each routed job, so a restrictive limit can leave some routed jobs
+ with no matching hosts, the same way it can with sliced jobs.
+- Relaunching a routed job (a single bucket) reruns the same bucket on the same
+ instance group. Relaunching the routed workflow recomputes the buckets from
+ the current inventory data.