diff --git a/awx/main/models/unified_jobs.py b/awx/main/models/unified_jobs.py
index 76696deac0dc..b8dd1d1cf857 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -42,9 +42,9 @@
     camelcase_to_underscore, get_model_for_type,
     encrypt_dict, decrypt_field, _inventory_updates,
     copy_model_by_class, copy_m2m_relationships,
-    get_type_for_model, parse_yaml_or_json, getattr_dne
+    get_type_for_model, parse_yaml_or_json, getattr_dne,
+    polymorphic, schedule_task_manager
 )
-from awx.main.utils import polymorphic, schedule_task_manager
 from awx.main.constants import ACTIVE_STATES, CAN_CANCEL
 from awx.main.redact import UriCleaner, REPLACE_STR
 from awx.main.consumers import emit_channel_notification
@@ -1386,9 +1386,13 @@ def awx_meta_vars(self):
 
         wj = self.get_workflow_job()
         if wj:
+            schedule = getattr_dne(wj, 'schedule')
             for name in ('awx', 'tower'):
                 r['{}_workflow_job_id'.format(name)] = wj.pk
                 r['{}_workflow_job_name'.format(name)] = wj.name
+                if schedule:
+                    r['{}_parent_job_schedule_id'.format(name)] = schedule.pk
+                    r['{}_parent_job_schedule_name'.format(name)] = schedule.name
 
         if not created_by:
             schedule = getattr_dne(self, 'schedule')
diff --git a/awx/main/tests/functional/models/test_unified_job.py b/awx/main/tests/functional/models/test_unified_job.py
index 0db76aac4c06..4d0418c4989e 100644
--- a/awx/main/tests/functional/models/test_unified_job.py
+++ b/awx/main/tests/functional/models/test_unified_job.py
@@ -147,6 +147,39 @@ def test_scheduled_job_metavars(self, job_template, admin_user):
         assert data['awx_schedule_id'] == schedule.pk
         assert 'awx_user_name' not in data
 
+    def test_scheduled_workflow_job_node_metavars(self, workflow_job_template):
+        schedule = Schedule.objects.create(
+            name='job-schedule',
+            rrule='DTSTART:20171129T155939z\nFREQ=MONTHLY',
+            unified_job_template=workflow_job_template
+        )
+
+        workflow_job = WorkflowJob.objects.create(
+            name='workflow-job',
+            workflow_job_template=workflow_job_template,
+            schedule=schedule
+        )
+
+        job = Job.objects.create(
+            launch_type='workflow'
+        )
+        workflow_job.workflow_nodes.create(job=job)
+        assert job.awx_meta_vars() == {
+            'awx_job_id': job.id,
+            'tower_job_id': job.id,
+            'awx_job_launch_type': 'workflow',
+            'tower_job_launch_type': 'workflow',
+            'awx_workflow_job_name': 'workflow-job',
+            'tower_workflow_job_name': 'workflow-job',
+            'awx_workflow_job_id': workflow_job.id,
+            'tower_workflow_job_id': workflow_job.id,
+            'awx_parent_job_schedule_id': schedule.id,
+            'tower_parent_job_schedule_id': schedule.id,
+            'awx_parent_job_schedule_name': 'job-schedule',
+            'tower_parent_job_schedule_name': 'job-schedule',
+            
+        }
+
 
 @pytest.mark.django_db
 def test_event_processing_not_finished():