Add canonical receptor address

Creates a non-deletable address that acts as
the "main" address for this instance.

All other addresses for that instance must
be non-canonical.

When listener_port on an instance is set, automatically
create a canonical receptor address where:
  - address is hostname of instance
  - port is listener_port
  - canonical is True

Additionally, protocol field is added to instance to
denote the receptor listener protocol to use (ws, tcp).

The receptor config listener information is derived from
the listener_port and protocol information. Having a
canonical address that mirrors the listener_port ensures that
an address exists that matches the receptor config information.

Other changes:
- Add managed field to receptor address.
If managed is True, no fields on on this address can be edited
via the API.
If canonical is True, only the address cannot be edited.

- Add managed field to instance. If managed is True, users
cannot set node_state to deprovisioning (i.e. cannot delete node)

This change to our mechanism to prevent users from deleting
the mesh ingress hop node.

- Field is_internal is now renamed to k8s_routable

- Add reverse_peers on instance which is a list of instance IDs
that peer to this instance (via an address)

Signed-off-by: Seth Foster <fosterbseth@gmail.com>

awx/api/serializers.py

@@ -5490,8 +5490,20 @@ class ReceptorAddressSerializer(BaseSerializer):
 
     class Meta:
         model = ReceptorAddress
-        fields = ('id', 'url', 'address', 'port', 'protocol', 'websocket_path', 'is_internal', 'instance', 'peers_from_control_nodes', 'full_address')
-        read_only_fields = ('full_address',)
+        fields = (
+            'id',
+            'url',
+            'address',
+            'port',
+            'websocket_path',
+            'k8s_routable',
+            'canonical',
+            'instance',
+            'managed',
+            'peers_from_control_nodes',
+            'full_address',
+        )
+        read_only_fields = ('full_address', 'managed', 'canonical', 'k8s_routable')
 
     def get_full_address(self, obj):
         return obj.get_full_address()
@@ -5500,25 +5512,47 @@ 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)
 
+        managed = get_field_from_model_or_attrs('managed')
+        canonical = get_field_from_model_or_attrs('canonical')
+
+        if managed:
+            raise serializers.ValidationError(_("Cannot modify a managed address."))
+
+        # cannot modify address field if canonical is True
+        if canonical and attrs.get('address') and self.instance and self.instance.address != attrs.get('address'):
+            raise serializers.ValidationError(_("Cannot modify address field if it is canonical."))
+
         peers_from_control_nodes = get_field_from_model_or_attrs('peers_from_control_nodes')
         instance = get_field_from_model_or_attrs('instance')
+        address = get_field_from_model_or_attrs('address')
+
+        if not instance.listener_port:
+            raise serializers.ValidationError(_("Instance must have a listener port set."))
 
-        # only allow websocket_path to be set if protocol is ws
-        if attrs.get('protocol') != 'ws' and attrs.get('websocket_path'):
+        # only allow websocket_path to be set if instance protocol is ws
+        if attrs.get('websocket_path') and instance and instance.protocol != 'ws':
             raise serializers.ValidationError(_("Can only set websocket path if protocol is ws."))
 
         # an instance can only have one address with peers_from_control_nodes set to True
         if peers_from_control_nodes:
             for other_address in ReceptorAddress.objects.filter(instance=instance.id):
-                if other_address.peers_from_control_nodes:
+                if other_address.address != address and other_address.peers_from_control_nodes:
                     raise serializers.ValidationError(_("Only one address can set peers_from_control_nodes to True."))
 
-        # is_internal should be False
-        if attrs.get('is_internal') == True:
+        # k8s_routable should be False
+        if attrs.get('k8s_routable') == True:
             raise serializers.ValidationError(_("Only external addresses can be created."))
 
         return super().validate(attrs)
 
+    def update(self, obj, validated_data):
+        addr = super(ReceptorAddressSerializer, self).update(obj, validated_data)
+        if addr.port != addr.instance.listener_port:
+            addr.instance.listener_port = addr.port
+            addr.instance.save(update_fields=['listener_port'])
+
+        return addr
+
 
 class InstanceSerializer(BaseSerializer):
     show_capabilities = ['edit']
@@ -5531,10 +5565,11 @@ class InstanceSerializer(BaseSerializer):
     peers = serializers.PrimaryKeyRelatedField(
         help_text=_('Primary keys of receptor addresses to peer to.'), many=True, required=False, queryset=ReceptorAddress.objects.all()
     )
+    reverse_peers = serializers.SerializerMethodField()
 
     class Meta:
         model = Instance
-        read_only_fields = ('ip_address', 'uuid', 'version')
+        read_only_fields = ('ip_address', 'uuid', 'version', 'managed', 'reverse_peers')
         fields = (
             'id',
             'hostname',
@@ -5565,8 +5600,12 @@ class Meta:
             'managed_by_policy',
             'node_type',
             'node_state',
+            'managed',
             'ip_address',
             'peers',
+            'reverse_peers',
+            'listener_port',
+            'protocol',
         )
         extra_kwargs = {
             'node_type': {'initial': Instance.Types.EXECUTION, 'default': Instance.Types.EXECUTION},
@@ -5599,6 +5638,37 @@ def get_related(self, obj):
                 res['health_check'] = self.reverse('api:instance_health_check', kwargs={'pk': obj.pk})
         return res
 
+    def create(self, validated_data):
+        # create a managed receptor address if listener port is defined
+        kwargs = {
+            'port': validated_data.get('listener_port', None),
+            'canonical': True,
+        }
+        kwargs = {k: v for k, v in kwargs.items() if v is not None}
+        instance = super(InstanceSerializer, self).create(validated_data)
+        if kwargs.get('port'):
+            instance.receptor_addresses.update_or_create(address=instance.hostname, defaults=kwargs)
+        else:
+            # delete the receptor address if the listener port is not defined
+            instance.receptor_addresses.filter(address=instance.hostname).delete()
+        return instance
+
+    def update(self, obj, validated_data):
+        # update the managed receptor address if listener port is defined
+        kwargs = {
+            'port': validated_data.get('listener_port', None),
+            'canonical': True,
+        }
+        kwargs = {k: v for k, v in kwargs.items() if v is not None}
+        instance = super(InstanceSerializer, self).update(obj, validated_data)
+        if kwargs.get('port'):
+            instance.receptor_addresses.update_or_create(address=instance.hostname, defaults=kwargs)
+        else:
+            # delete the receptor address if the listener port is not defined
+            instance.receptor_addresses.filter(address=instance.hostname).delete()
+
+        return instance
+
     def get_summary_fields(self, obj):
         summary = super().get_summary_fields(obj)
 
@@ -5608,6 +5678,9 @@ def get_summary_fields(self, obj):
 
         return summary
 
+    def get_reverse_peers(self, obj):
+        return Instance.objects.prefetch_related('peers').filter(peers__in=obj.receptor_addresses.all()).values_list('id', flat=True)
+
     def get_consumed_capacity(self, obj):
         return obj.consumed_capacity
 
@@ -5683,8 +5756,8 @@ def validate_node_state(self, value):
                     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, Instance.Types.HOP):
-                    raise serializers.ValidationError(_("Can only deprovision execution or hop nodes."))
+                if self.instance.managed:
+                    raise serializers.ValidationError(_("Cannot deprovision managed nodes."))
         else:
             if value and value != Instance.States.INSTALLED:
                 raise serializers.ValidationError(_("Can only create instances in the 'installed' state."))

awx/api/views/__init__.py

@@ -405,6 +405,12 @@ class ReceptorAddressDetail(RetrieveUpdateDestroyAPIView):
     model = models.ReceptorAddress
     serializer_class = serializers.ReceptorAddressSerializer
 
+    def delete(self, request, *args, **kwargs):
+        obj = self.get_object()
+        if obj.canonical or obj.managed:
+            return Response({'detail': _('Cannot delete canonical or managed address.')}, status=status.HTTP_400_BAD_REQUEST)
+        return super(ReceptorAddressDetail, self).delete(request, *args, **kwargs)
+
 
 class InstanceInstanceGroupsList(InstanceGroupMembershipMixin, SubListCreateAttachDetachAPIView):
     name = _("Instance's Instance Groups")

awx/api/views/instance_install_bundle.py

@@ -126,15 +126,12 @@ def generate_inventory_yml(instance_obj):
 def generate_group_vars_all_yml(instance_obj):
     # get peers
     peers = []
-    for addr in instance_obj.peers.all():
-        peers.append(dict(address=addr.get_full_address(), protocol=addr.protocol))
+    for addr in instance_obj.peers.all().prefetch_related('instance'):
+        peers.append(dict(address=addr.get_full_address(), protocol=addr.instance.protocol))
     context = dict(instance=instance_obj, peers=peers)
 
-    # we infer the listener port information from the first tcp receptor address
-    # currently for external remote nodes, we only support a single tcp backend listeners
-    listener_addr = instance_obj.receptor_addresses.filter(protocol="tcp").first()
-    if listener_addr:
-        context['listener_port'] = listener_addr.port
+    if instance_obj.listener_port:
+        context['listener_port'] = instance_obj.listener_port
 
     all_yaml = render_to_string("instance_install_bundle/group_vars/all.yml", context=context)
     # convert consecutive newlines with a single newline

awx/main/management/commands/add_receptor_address.py

@@ -8,13 +8,16 @@
 
 def add_address(**kwargs):
     try:
-        instance = Instance.objects.get(hostname=kwargs.pop('hostname'))
+        instance = Instance.objects.get(hostname=kwargs.pop('instance'))
         kwargs['instance'] = instance
-        # address and protocol are unique together for ReceptorAddress
-        # If an address has (address, protocol), it will update the rest of the values suppled in defaults dict
-        # if no address exists with (address, protocol), then a new address will be created
-        # these unique together fields need to be consistent with the unique constraint in the ReceptorAddress model
-        addr, _ = ReceptorAddress.objects.update_or_create(address=kwargs.pop('address'), protocol=kwargs.pop('protocol'), defaults=kwargs)
+        # if ReceptorAddress already exists with address, just update
+        # otherwise, create new ReceptorAddress
+        addr, _ = ReceptorAddress.objects.update_or_create(address=kwargs.pop('address'), defaults=kwargs)
+
+        # update listener_port on instance if address is canonical
+        if addr.canonical:
+            addr.instance.listener_port = addr.port
+            addr.instance.save(update_fields=['listener_port'])
         print(f"Successfully added receptor address {addr.get_full_address()}")
         changed = True
     except Exception as e:
@@ -32,17 +35,22 @@ class Command(BaseCommand):
     help = "Add receptor address to an instance."
 
     def add_arguments(self, parser):
-        parser.add_argument('--hostname', dest='hostname', type=str, help="Hostname this address is added to")
+        parser.add_argument('--instance', dest='instance', type=str, help="Instance hostname this address is added to")
         parser.add_argument('--address', dest='address', type=str, help="Receptor address")
         parser.add_argument('--port', dest='port', type=int, help="Receptor listener port")
-        parser.add_argument('--protocol', dest='protocol', type=str, default='tcp', choices=['tcp', 'ws'], help="Protocol of the backend connection")
         parser.add_argument('--websocket_path', dest='websocket_path', type=str, default="", help="Path for websockets")
-        parser.add_argument('--is_internal', action='store_true', help="If true, address only resolvable within the Kubernetes cluster")
+        parser.add_argument('--k8s_routable', action='store_true', help="If true, address only resolvable within the Kubernetes cluster")
+        parser.add_argument('--canonical', action='store_true', help="If true, address is the canonical address for the instance")
         parser.add_argument('--peers_from_control_nodes', action='store_true', help="If true, control nodes will peer to this address")
+        parser.add_argument('--managed', action='store_true', help="If True, this address should be managed by the control plane.")
 
     def handle(self, **options):
         self.changed = False
-        address_options = {k: options[k] for k in ('hostname', 'address', 'port', 'protocol', 'websocket_path', 'is_internal', 'peers_from_control_nodes')}
+        address_options = {
+            k: options[k]
+            for k in ('instance', 'address', 'port', 'websocket_path', 'k8s_routable', 'peers_from_control_nodes', 'canonical', 'managed')
+            if options[k]
+        }
         self.changed = add_address(**address_options)
         if self.changed:
             print("(changed: True)")

awx/main/management/commands/provision_instance.py

@@ -26,18 +26,21 @@ class Command(BaseCommand):
     def add_arguments(self, parser):
         parser.add_argument('--hostname', dest='hostname', type=str, help="Hostname used during provisioning")
         parser.add_argument('--listener_port', dest='listener_port', type=int, help="Receptor listener port")
+        parser.add_argument(
+            '--protocol', dest='protocol', type=str, default='tcp', choices=['tcp', 'ws', 'wss'], help="Protocol to use for the Receptor listener"
+        )
         parser.add_argument('--node_type', type=str, default='hybrid', choices=['control', 'execution', 'hop', 'hybrid'], help="Instance Node type")
         parser.add_argument('--uuid', type=str, help="Instance UUID")
 
-    def _register_hostname(self, hostname, node_type, uuid, listener_port):
+    def _register_hostname(self, hostname, node_type, uuid, listener_port, protocol):
         if not hostname:
             if not settings.AWX_AUTO_DEPROVISION_INSTANCES:
                 raise CommandError('Registering with values from settings only intended for use in K8s installs')
 
             from awx.main.management.commands.register_queue import RegisterQueue
 
             (changed, instance) = Instance.objects.register(
-                ip_address=os.environ.get('MY_POD_IP'), listener_port=listener_port, node_type='control', node_uuid=settings.SYSTEM_UUID
+                ip_address=os.environ.get('MY_POD_IP'), listener_port=listener_port, node_type='control', node_uuid=settings.SYSTEM_UUID, protocol=protocol
             )
             RegisterQueue(settings.DEFAULT_CONTROL_PLANE_QUEUE_NAME, 100, 0, [], is_container_group=False).register()
             RegisterQueue(
@@ -51,16 +54,19 @@ def _register_hostname(self, hostname, node_type, uuid, listener_port):
                 max_concurrent_jobs=settings.DEFAULT_EXECUTION_QUEUE_MAX_CONCURRENT_JOBS,
             ).register()
         else:
-            (changed, instance) = Instance.objects.register(hostname=hostname, node_type=node_type, node_uuid=uuid, listener_port=listener_port)
+            (changed, instance) = Instance.objects.register(
+                hostname=hostname, node_type=node_type, node_uuid=uuid, listener_port=listener_port, protocol=protocol
+            )
         if changed:
             print("Successfully registered instance {}".format(hostname))
         else:
             print("Instance already registered {}".format(instance.hostname))
+
         self.changed = changed
 
     @transaction.atomic
     def handle(self, **options):
         self.changed = False
-        self._register_hostname(options.get('hostname'), options.get('node_type'), options.get('uuid'), options.get('listener_port'))
+        self._register_hostname(options.get('hostname'), options.get('node_type'), options.get('uuid'), options.get('listener_port'), options.get('protocol'))
         if self.changed:
             print("(changed: True)")

awx/main/managers.py

@@ -115,7 +115,17 @@ def me(self):
             return node[0]
         raise RuntimeError("No instance found with the current cluster host id")
 
-    def register(self, node_uuid=None, hostname=None, ip_address="", listener_port=None, node_type='hybrid', peers_from_control_nodes=False, defaults=None):
+    def register(
+        self,
+        node_uuid=None,
+        hostname=None,
+        ip_address="",
+        listener_port=None,
+        protocol='tcp',
+        node_type='hybrid',
+        peers_from_control_nodes=False,
+        defaults=None,
+    ):
         if not hostname:
             hostname = settings.CLUSTER_HOST_ID
 
@@ -161,6 +171,12 @@ def register(self, node_uuid=None, hostname=None, ip_address="", listener_port=N
                 if instance.node_type != node_type:
                     instance.node_type = node_type
                     update_fields.append('node_type')
+                if instance.protocol != protocol:
+                    instance.protocol = protocol
+                    update_fields.append('protocol')
+                if instance.listener_port != listener_port:
+                    instance.listener_port = listener_port
+                    update_fields.append('listener_port')
                 if update_fields:
                     instance.save(update_fields=update_fields)
                     return (True, instance)
@@ -171,6 +187,7 @@ def register(self, node_uuid=None, hostname=None, ip_address="", listener_port=N
             create_defaults = {
                 'node_state': Instance.States.INSTALLED,
                 'capacity': 0,
+                'managed': True,
             }
             if defaults is not None:
                 create_defaults.update(defaults)
@@ -185,8 +202,5 @@ def register(self, node_uuid=None, hostname=None, ip_address="", listener_port=N
                 **create_defaults,
                 **uuid_option
             )
-            from awx.main.management.commands.add_receptor_address import add_address
 
-            if listener_port:
-                add_address(address=hostname, hostname=hostname, port=listener_port, protocol='tcp')
         return (True, instance)