Hop nodes for k8s

awx/api/serializers.py

@@ -5356,10 +5356,16 @@ def validate(self, attrs):
 class InstanceLinkSerializer(BaseSerializer):
     class Meta:
         model = InstanceLink
-        fields = ('source', 'target', 'link_state')
+        fields = ('id', 'url', 'related', 'source', 'target', 'link_state')
 
-    source = serializers.SlugRelatedField(slug_field="hostname", read_only=True)
-    target = serializers.SlugRelatedField(slug_field="hostname", read_only=True)
+    source = serializers.SlugRelatedField(slug_field="hostname", queryset=Instance.objects.all())
+    target = serializers.SlugRelatedField(slug_field="hostname", queryset=Instance.objects.all())
+
+    def get_related(self, obj):
+        res = super(InstanceLinkSerializer, self).get_related(obj)
+        res['source_instance'] = self.reverse('api:instance_detail', kwargs={'pk': obj.source.id})
+        res['target_instance'] = self.reverse('api:instance_detail', kwargs={'pk': obj.target.id})
+        return res
 
 
 class InstanceNodeSerializer(BaseSerializer):
@@ -5376,6 +5382,7 @@ class InstanceSerializer(BaseSerializer):
     jobs_running = serializers.IntegerField(help_text=_('Count of jobs in the running or waiting state that are targeted for this instance'), read_only=True)
     jobs_total = serializers.IntegerField(help_text=_('Count of all jobs that target this instance'), read_only=True)
     health_check_pending = serializers.SerializerMethodField()
+    peers = serializers.SlugRelatedField(many=True, required=False, slug_field="hostname", queryset=Instance.objects.all())
 
     class Meta:
         model = Instance
@@ -5412,6 +5419,8 @@ class Meta:
             'node_state',
             'ip_address',
             'listener_port',
+            'peers',
+            'peers_from_control_nodes',
         )
         extra_kwargs = {
             'node_type': {'initial': Instance.Types.EXECUTION, 'default': Instance.Types.EXECUTION},
@@ -5464,61 +5473,119 @@ def get_percent_capacity_remaining(self, obj):
     def get_health_check_pending(self, obj):
         return obj.health_check_pending
 
-    def validate(self, data):
-        if self.instance:
-            if self.instance.node_type == Instance.Types.HOP:
-                raise serializers.ValidationError("Hop node instances may not be changed.")
-        else:
-            if not settings.IS_K8S:
-                raise serializers.ValidationError("Can only create instances on Kubernetes or OpenShift.")
-        return data
+    def validate(self, attrs):
+        def get_field_from_model_or_attrs(fd):
+            return attrs.get(fd, self.instance and getattr(self.instance, fd) or None)
+
+        def check_peers_changed():
+            '''
+            return True if
+            - 'peers' in attrs
+            - instance peers matches peers in attrs
+            '''
+            return self.instance and 'peers' in attrs and set(self.instance.peers.all()) != set(attrs['peers'])
+
+        if not self.instance and not settings.IS_K8S:
+            raise serializers.ValidationError(_("Can only create instances on Kubernetes or OpenShift."))
+
+        node_type = get_field_from_model_or_attrs("node_type")
+        peers_from_control_nodes = get_field_from_model_or_attrs("peers_from_control_nodes")
+        listener_port = get_field_from_model_or_attrs("listener_port")
+        peers = attrs.get('peers', [])
+
+        if peers_from_control_nodes and node_type not in (Instance.Types.EXECUTION, Instance.Types.HOP):
+            raise serializers.ValidationError(_("peers_from_control_nodes can only be enabled for execution or hop nodes."))
+
+        if node_type in [Instance.Types.CONTROL, Instance.Types.HYBRID]:
+            if check_peers_changed():
+                raise serializers.ValidationError(
+                    _("Setting peers manually for control nodes is not allowed. Enable peers_from_control_nodes on the hop and execution nodes instead.")
+                )
+
+        if not listener_port and peers_from_control_nodes:
+            raise serializers.ValidationError(_("Field listener_port must be a valid integer when peers_from_control_nodes is enabled."))
+
+        if not listener_port and self.instance and self.instance.peers_from.exists():
+            raise serializers.ValidationError(_("Field listener_port must be a valid integer when other nodes peer to it."))
+
+        for peer in peers:
+            if peer.listener_port is None:
+                raise serializers.ValidationError(_("Field listener_port must be set on peer ") + peer.hostname + ".")
+
+        if not settings.IS_K8S:
+            if check_peers_changed():
+                raise serializers.ValidationError(_("Cannot change peers."))
+
+        return super().validate(attrs)
 
     def validate_node_type(self, value):
-        if not self.instance:
-            if value not in (Instance.Types.EXECUTION,):
-                raise serializers.ValidationError("Can only create execution nodes.")
-        else:
-            if self.instance.node_type != value:
-                raise serializers.ValidationError("Cannot change node type.")
+        if not self.instance and value not in [Instance.Types.HOP, Instance.Types.EXECUTION]:
+            raise serializers.ValidationError(_("Can only create execution or hop nodes."))
+
+        if self.instance and self.instance.node_type != value:
+            raise serializers.ValidationError(_("Cannot change node type."))
 
         return value
 
     def validate_node_state(self, value):
         if self.instance:
             if value != self.instance.node_state:
                 if not settings.IS_K8S:
-                    raise serializers.ValidationError("Can only change the state on Kubernetes or OpenShift.")
+                    raise serializers.ValidationError(_("Can only change the state on Kubernetes or OpenShift."))
                 if value != Instance.States.DEPROVISIONING:
-                    raise serializers.ValidationError("Can only change instances to the 'deprovisioning' state.")
-                if self.instance.node_type not in (Instance.Types.EXECUTION,):
-                    raise serializers.ValidationError("Can only deprovision execution nodes.")
+                    raise serializers.ValidationError(_("Can only change instances to the 'deprovisioning' state."))
+                if self.instance.node_type not in (Instance.Types.EXECUTION, Instance.Types.HOP):
+                    raise serializers.ValidationError(_("Can only deprovision execution or hop nodes."))
         else:
             if value and value != Instance.States.INSTALLED:
-                raise serializers.ValidationError("Can only create instances in the 'installed' state.")
+                raise serializers.ValidationError(_("Can only create instances in the 'installed' state."))
 
         return value
 
     def validate_hostname(self, value):
         """
-        - Hostname cannot be "localhost" - but can be something like localhost.domain
-        - Cannot change the hostname of an-already instantiated & initialized Instance object
+        Cannot change the hostname
         """
         if self.instance and self.instance.hostname != value:
-            raise serializers.ValidationError("Cannot change hostname.")
+            raise serializers.ValidationError(_("Cannot change hostname."))
 
         return value
 
     def validate_listener_port(self, value):
-        if self.instance and self.instance.listener_port != value:
-            raise serializers.ValidationError("Cannot change listener port.")
+        """
+        Cannot change listener port, unless going from none to integer, and vice versa
+        """
+        if value and self.instance and self.instance.listener_port and self.instance.listener_port != value:
+            raise serializers.ValidationError(_("Cannot change listener port."))
+
+        return value
+
+    def validate_peers_from_control_nodes(self, value):
+        """
+        Can only enable for K8S based deployments
+        """
+        if value and not settings.IS_K8S:
+            raise serializers.ValidationError(_("Can only be enabled on Kubernetes or Openshift."))
 
         return value
 
 
 class InstanceHealthCheckSerializer(BaseSerializer):
     class Meta:
         model = Instance
-        read_only_fields = ('uuid', 'hostname', 'version', 'last_health_check', 'errors', 'cpu', 'memory', 'cpu_capacity', 'mem_capacity', 'capacity')
+        read_only_fields = (
+            'uuid',
+            'hostname',
+            'ip_address',
+            'version',
+            'last_health_check',
+            'errors',
+            'cpu',
+            'memory',
+            'cpu_capacity',
+            'mem_capacity',
+            'capacity',
+        )
         fields = read_only_fields
 
 

awx/api/templates/instance_install_bundle/group_vars/all.yml

@@ -3,21 +3,35 @@ receptor_group: awx
 receptor_verify: true
 receptor_tls: true
 receptor_mintls13: false
+{% if instance.node_type == "execution" %}
 receptor_work_commands:
   ansible-runner:
     command: ansible-runner
     params: worker
     allowruntimeparams: true
     verifysignature: true
+additional_python_packages:
+  - ansible-runner
+{% endif %}
 custom_worksign_public_keyfile: receptor/work_public_key.pem
 custom_tls_certfile: receptor/tls/receptor.crt
 custom_tls_keyfile: receptor/tls/receptor.key
 custom_ca_certfile: receptor/tls/ca/mesh-CA.crt
 receptor_protocol: 'tcp'
+{% if instance.listener_port %}
 receptor_listener: true
 receptor_port: {{ instance.listener_port }}
-receptor_dependencies:
-  - python39-pip
+{% else %}
+receptor_listener: false
+{% endif %}
+{% if peers %}
+receptor_peers:
+{% for peer in peers %}
+  - host: {{ peer.host }}
+    port: {{ peer.port }}
+    protocol: tcp
+{% endfor %}
+{% endif %}
 {% verbatim %}
 podman_user: "{{ receptor_user }}"
 podman_group: "{{ receptor_group }}"

awx/api/templates/instance_install_bundle/install_receptor.yml

@@ -1,20 +1,16 @@
-{% verbatim %}
 ---
 - hosts: all
   become: yes
   tasks:
     - name: Create the receptor user
       user:
+{% verbatim %}
         name: "{{ receptor_user }}"
+{% endverbatim %}
         shell: /bin/bash
-    - name: Enable Copr repo for Receptor
-      command: dnf copr enable ansible-awx/receptor -y
+{% if instance.node_type == "execution" %}
     - import_role:
         name: ansible.receptor.podman
+{% endif %}
     - import_role:
         name: ansible.receptor.setup
-    - name: Install ansible-runner
-      pip:
-        name: ansible-runner
-        executable: pip3.9
-{% endverbatim %}

awx/api/templates/instance_install_bundle/requirements.yml

@@ -1,4 +1,5 @@
 ---
 collections:
-  - name: ansible.receptor
-    version: 1.1.0
+  - name: https://github.com/ansible/receptor-collection.git
+    type: git
+    version: main

awx/api/views/__init__.py

@@ -341,17 +341,18 @@ class InstanceDetail(RetrieveUpdateAPIView):
 
     def update_raw_data(self, data):
         # these fields are only valid on creation of an instance, so they unwanted on detail view
-        data.pop('listener_port', None)
         data.pop('node_type', None)
         data.pop('hostname', None)
+        data.pop('ip_address', None)
         return super(InstanceDetail, self).update_raw_data(data)
 
     def update(self, request, *args, **kwargs):
         r = super(InstanceDetail, self).update(request, *args, **kwargs)
         if status.is_success(r.status_code):
             obj = self.get_object()
-            obj.set_capacity_value()
-            obj.save(update_fields=['capacity'])
+            capacity_changed = obj.set_capacity_value()
+            if capacity_changed:
+                obj.save(update_fields=['capacity'])
             r.data = serializers.InstanceSerializer(obj, context=self.get_serializer_context()).to_representation(obj)
         return r
 

awx/api/views/instance_install_bundle.py

@@ -6,6 +6,8 @@
 import ipaddress
 import os
 import tarfile
+import time
+import re
 
 import asn1
 from awx.api import serializers
@@ -40,6 +42,8 @@
 # │   │   └── receptor.key
 # │   └── work-public-key.pem
 # └── requirements.yml
+
+
 class InstanceInstallBundle(GenericAPIView):
     name = _('Install Bundle')
     model = models.Instance
@@ -49,9 +53,9 @@ class InstanceInstallBundle(GenericAPIView):
     def get(self, request, *args, **kwargs):
         instance_obj = self.get_object()
 
-        if instance_obj.node_type not in ('execution',):
+        if instance_obj.node_type not in ('execution', 'hop'):
             return Response(
-                data=dict(msg=_('Install bundle can only be generated for execution nodes.')),
+                data=dict(msg=_('Install bundle can only be generated for execution or hop nodes.')),
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
@@ -66,37 +70,37 @@ def get(self, request, *args, **kwargs):
                 # generate and write the receptor key to receptor/tls/receptor.key in the tar file
                 key, cert = generate_receptor_tls(instance_obj)
 
+                def tar_addfile(tarinfo, filecontent):
+                    tarinfo.mtime = time.time()
+                    tarinfo.size = len(filecontent)
+                    tar.addfile(tarinfo, io.BytesIO(filecontent))
+
                 key_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/receptor/tls/receptor.key")
-                key_tarinfo.size = len(key)
-                tar.addfile(key_tarinfo, io.BytesIO(key))
+                tar_addfile(key_tarinfo, key)
 
                 cert_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/receptor/tls/receptor.crt")
                 cert_tarinfo.size = len(cert)
-                tar.addfile(cert_tarinfo, io.BytesIO(cert))
+                tar_addfile(cert_tarinfo, cert)
 
                 # generate and write install_receptor.yml to the tar file
-                playbook = generate_playbook().encode('utf-8')
+                playbook = generate_playbook(instance_obj).encode('utf-8')
                 playbook_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/install_receptor.yml")
-                playbook_tarinfo.size = len(playbook)
-                tar.addfile(playbook_tarinfo, io.BytesIO(playbook))
+                tar_addfile(playbook_tarinfo, playbook)
 
                 # generate and write inventory.yml to the tar file
                 inventory_yml = generate_inventory_yml(instance_obj).encode('utf-8')
                 inventory_yml_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/inventory.yml")
-                inventory_yml_tarinfo.size = len(inventory_yml)
-                tar.addfile(inventory_yml_tarinfo, io.BytesIO(inventory_yml))
+                tar_addfile(inventory_yml_tarinfo, inventory_yml)
 
                 # generate and write group_vars/all.yml to the tar file
                 group_vars = generate_group_vars_all_yml(instance_obj).encode('utf-8')
                 group_vars_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/group_vars/all.yml")
-                group_vars_tarinfo.size = len(group_vars)
-                tar.addfile(group_vars_tarinfo, io.BytesIO(group_vars))
+                tar_addfile(group_vars_tarinfo, group_vars)
 
                 # generate and write requirements.yml to the tar file
                 requirements_yml = generate_requirements_yml().encode('utf-8')
                 requirements_yml_tarinfo = tarfile.TarInfo(f"{instance_obj.hostname}_install_bundle/requirements.yml")
-                requirements_yml_tarinfo.size = len(requirements_yml)
-                tar.addfile(requirements_yml_tarinfo, io.BytesIO(requirements_yml))
+                tar_addfile(requirements_yml_tarinfo, requirements_yml)
 
             # respond with the tarfile
             f.seek(0)
@@ -105,8 +109,10 @@ def get(self, request, *args, **kwargs):
             return response
 
 
-def generate_playbook():
-    return render_to_string("instance_install_bundle/install_receptor.yml")
+def generate_playbook(instance_obj):
+    playbook_yaml = render_to_string("instance_install_bundle/install_receptor.yml", context=dict(instance=instance_obj))
+    # convert consecutive newlines with a single newline
+    return re.sub(r'\n+', '\n', playbook_yaml)
 
 
 def generate_requirements_yml():
@@ -118,7 +124,12 @@ def generate_inventory_yml(instance_obj):
 
 
 def generate_group_vars_all_yml(instance_obj):
-    return render_to_string("instance_install_bundle/group_vars/all.yml", context=dict(instance=instance_obj))
+    peers = []
+    for instance in instance_obj.peers.all():
+        peers.append(dict(host=instance.hostname, port=instance.listener_port))
+    all_yaml = render_to_string("instance_install_bundle/group_vars/all.yml", context=dict(instance=instance_obj, peers=peers))
+    # convert consecutive newlines with a single newline
+    return re.sub(r'\n+', '\n', all_yaml)
 
 
 def generate_receptor_tls(instance_obj):