path: root/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/routing/RoutingPolicies.java

// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.hosted.controller.routing;

import ai.vespa.http.DomainName;
import com.yahoo.config.application.api.DeploymentSpec;
import com.yahoo.config.provision.ApplicationId;
import com.yahoo.config.provision.ClusterSpec;
import com.yahoo.config.provision.zone.AuthMethod;
import com.yahoo.config.provision.zone.RoutingMethod;
import com.yahoo.config.provision.zone.ZoneId;
import com.yahoo.transaction.Mutex;
import com.yahoo.vespa.hosted.controller.Application;
import com.yahoo.vespa.hosted.controller.Controller;
import com.yahoo.vespa.hosted.controller.api.identifiers.ClusterId;
import com.yahoo.vespa.hosted.controller.api.identifiers.DeploymentId;
import com.yahoo.vespa.hosted.controller.api.integration.configserver.LoadBalancer;
import com.yahoo.vespa.hosted.controller.api.integration.dns.AliasTarget;
import com.yahoo.vespa.hosted.controller.api.integration.dns.DirectTarget;
import com.yahoo.vespa.hosted.controller.api.integration.dns.LatencyAliasTarget;
import com.yahoo.vespa.hosted.controller.api.integration.dns.Record;
import com.yahoo.vespa.hosted.controller.api.integration.dns.RecordData;
import com.yahoo.vespa.hosted.controller.api.integration.dns.RecordName;
import com.yahoo.vespa.hosted.controller.api.integration.dns.VpcEndpointService.ChallengeState;
import com.yahoo.vespa.hosted.controller.api.integration.dns.VpcEndpointService.DnsChallenge;
import com.yahoo.vespa.hosted.controller.api.integration.dns.WeightedAliasTarget;
import com.yahoo.vespa.hosted.controller.api.integration.dns.WeightedDirectTarget;
import com.yahoo.vespa.hosted.controller.application.Endpoint;
import com.yahoo.vespa.hosted.controller.application.EndpointId;
import com.yahoo.vespa.hosted.controller.application.EndpointList;
import com.yahoo.vespa.hosted.controller.application.GeneratedEndpoint;
import com.yahoo.vespa.hosted.controller.application.TenantAndApplicationId;
import com.yahoo.vespa.hosted.controller.dns.NameServiceForwarder;
import com.yahoo.vespa.hosted.controller.dns.NameServiceQueue.Priority;
import com.yahoo.vespa.hosted.controller.dns.NameServiceRequest;
import com.yahoo.vespa.hosted.controller.persistence.CuratorDb;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;

/**
 * Updates routing policies and their associated DNS records based on a deployment's load balancers.
 *
 * @author mortent
 * @author mpolden
 */
public class RoutingPolicies {

    private static final Logger LOG = Logger.getLogger(RoutingPolicies.class.getName());

    private final Controller controller;
    private final CuratorDb db;

    public RoutingPolicies(Controller controller) {
        this.controller = Objects.requireNonNull(controller, "controller must be non-null");
        this.db = controller.curator();
        try (var lock = db.lockRoutingPolicies()) { // Update serialized format
            for (var policy : db.readRoutingPolicies().entrySet()) {
                db.writeRoutingPolicies(policy.getKey(), policy.getValue());
            }
        }
    }

    /** Read all routing policies for given deployment */
    public RoutingPolicyList read(DeploymentId deployment) {
        return read(deployment.applicationId()).deployment(deployment);
    }

    /** Read all routing policies for given instance */
    public RoutingPolicyList read(ApplicationId instance) {
        return RoutingPolicyList.copyOf(db.readRoutingPolicies(instance));
    }

    /** Read all routing policies for given application */
    public RoutingPolicyList read(TenantAndApplicationId application) {
        return db.readRoutingPolicies((instance) -> TenantAndApplicationId.from(instance).equals(application))
                 .values()
                 .stream()
                 .flatMap(Collection::stream)
                 .collect(Collectors.collectingAndThen(Collectors.toList(), RoutingPolicyList::copyOf));
    }

    /** Read all routing policies */
    private RoutingPolicyList readAll() {
        return db.readRoutingPolicies()
                 .values()
                 .stream()
                 .flatMap(Collection::stream)
                 .collect(Collectors.collectingAndThen(Collectors.toList(), RoutingPolicyList::copyOf));
    }

    /** Read routing policy for given zone */
    public ZoneRoutingPolicy read(ZoneId zone) {
        return db.readZoneRoutingPolicy(zone);
    }

    /**
     * Refresh routing policies for instance in given zone. This is idempotent and changes will only be performed if
     * routing configuration affecting given deployment has changed.
     */
    public void refresh(DeploymentId deployment, DeploymentSpec deploymentSpec, EndpointList generatedEndpoints) {
        if (!generatedEndpoints.not().generated().isEmpty()) {
            throw new IllegalStateException("Generated endpoints contains non-generated, got " + generatedEndpoints);
        }
        ApplicationId instance = deployment.applicationId();
        List<LoadBalancer> loadBalancers = controller.serviceRegistry().configServer()
                                                     .getLoadBalancers(instance, deployment.zoneId());
        LoadBalancerAllocation allocation = new LoadBalancerAllocation(deployment, deploymentSpec, loadBalancers);
        Optional<TenantAndApplicationId> owner = ownerOf(allocation);
        try (var lock = db.lockRoutingPolicies()) {
            RoutingPolicyList applicationPolicies = read(TenantAndApplicationId.from(instance));
            RoutingPolicyList deploymentPolicies = applicationPolicies.deployment(allocation.deployment);

            removeGlobalDnsUnreferencedBy(allocation, deploymentPolicies, lock);
            removeApplicationDnsUnreferencedBy(allocation, deploymentPolicies, lock);

            RoutingPolicyList instancePolicies = storePoliciesOf(allocation, applicationPolicies, generatedEndpoints, lock);
            instancePolicies = removePoliciesUnreferencedBy(allocation, instancePolicies, lock);

            RoutingPolicyList updatedApplicationPolicies = applicationPolicies.replace(instance, instancePolicies);
            updateGlobalDnsOf(instancePolicies, Optional.of(deployment), owner, lock);
            updateApplicationDnsOf(updatedApplicationPolicies, deployment, owner, lock);
        }
    }

    /** Set the status of all global endpoints in given zone */
    public void setRoutingStatus(ZoneId zone, RoutingStatus.Value value) {
        try (var lock = db.lockRoutingPolicies()) {
            db.writeZoneRoutingPolicy(new ZoneRoutingPolicy(zone, RoutingStatus.create(value, RoutingStatus.Agent.operator,
                                                                                       controller.clock().instant())));
            Map<ApplicationId, RoutingPolicyList> allPolicies = readAll().groupingBy(policy -> policy.id().owner());
            allPolicies.forEach((instance, policies) -> {
                updateGlobalDnsOf(policies, Optional.empty(), Optional.of(TenantAndApplicationId.from(instance)), lock);
            });
        }
    }

    /** Set the status of all global endpoints for given deployment */
    public void setRoutingStatus(DeploymentId deployment, RoutingStatus.Value value, RoutingStatus.Agent agent) {
        ApplicationId instance = deployment.applicationId();
        try (var lock = db.lockRoutingPolicies()) {
            RoutingPolicyList applicationPolicies = read(TenantAndApplicationId.from(instance));
            RoutingPolicyList deploymentPolicies = applicationPolicies.deployment(deployment);
            Map<RoutingPolicyId, RoutingPolicy> updatedPolicies = new LinkedHashMap<>(applicationPolicies.asMap());
            for (var policy : deploymentPolicies) {
                var newPolicy = policy.with(RoutingStatus.create(value, agent, controller.clock().instant()));
                updatedPolicies.put(policy.id(), newPolicy);
            }
            RoutingPolicyList effectivePolicies = RoutingPolicyList.copyOf(updatedPolicies.values());
            Map<ApplicationId, RoutingPolicyList> policiesByInstance = effectivePolicies.groupingBy(policy -> policy.id().owner());
            policiesByInstance.forEach((ignored, instancePolicies) -> updateGlobalDnsOf(instancePolicies,
                                                                                        Optional.of(deployment),
                                                                                        ownerOf(deployment),
                                                                                        lock));
            updateApplicationDnsOf(effectivePolicies, deployment, ownerOf(deployment), lock);
            policiesByInstance.forEach((owner, instancePolicies) -> db.writeRoutingPolicies(owner, instancePolicies.asList()));
        }
    }

    /** Update global DNS records for given policies */
    private void updateGlobalDnsOf(RoutingPolicyList instancePolicies, Optional<DeploymentId> deployment,
                                   Optional<TenantAndApplicationId> owner,
                                   @SuppressWarnings("unused") Mutex lock) {
        Map<RoutingId, List<RoutingPolicy>> routingTable = instancePolicies.asInstanceRoutingTable();
        for (Map.Entry<RoutingId, List<RoutingPolicy>> routeEntry : routingTable.entrySet()) {
            RoutingId routingId = routeEntry.getKey();
            controller.routing().readDeclaredEndpointsOf(routingId.instance())
                      .named(routingId.endpointId(), Endpoint.Scope.global)
                      .not().requiresRotation()
                      .forEach(endpoint -> updateGlobalDnsOf(endpoint, routeEntry.getValue(), deployment, owner));
        }
    }

    /** Update global DNS records for given global endpoint */
    private void updateGlobalDnsOf(Endpoint endpoint, List<RoutingPolicy> policies,
                                   Optional<DeploymentId> deployment, Optional<TenantAndApplicationId> owner) {
        if (endpoint.scope() != Endpoint.Scope.global) throw new IllegalStateException("Endpoint " + endpoint + " is not global");
        if (deployment.isPresent() && !endpoint.deployments().contains(deployment.get())) return;

        Collection<RegionEndpoint> regionEndpoints = computeRegionEndpoints(endpoint, policies);
        Set<AliasTarget> latencyTargets = new LinkedHashSet<>();
        Set<AliasTarget> inactiveLatencyTargets = new LinkedHashSet<>();
        for (var regionEndpoint : regionEndpoints) {
            if (regionEndpoint.active()) {
                latencyTargets.add(regionEndpoint.target());
            } else {
                inactiveLatencyTargets.add(regionEndpoint.target());
            }
        }

        // Refuse removal of last target in an endpoint. We do this because removing 100% of the ALIAS records would
        // cause the application endpoint to stop resolving entirely (NXDOMAIN).
        if (latencyTargets.isEmpty() && !inactiveLatencyTargets.isEmpty()) {
            if (deployment.isPresent()) {
                throw new IllegalArgumentException("Cannot deactivate routing for " + deployment.get() +
                                                   " as it's the last remaining active deployment in " + endpoint);
            } else {
                // Operator is deactivating routing for entire zone, but this endpoint only has one target
                LOG.log(Level.WARNING, "Cannot deactivate routing for " + endpoint + " because it has only one " +
                                       "active zone. Leaving it in");
                return;
            }
        }

        // Create a weighted ALIAS per region, pointing to all zones within the same region
        regionEndpoints.forEach(regionEndpoint -> {
            if ( ! regionEndpoint.zoneAliasTargets().isEmpty()) {
                controller.nameServiceForwarder().createAlias(RecordName.from(regionEndpoint.target().name().value()),
                                                              regionEndpoint.zoneAliasTargets(),
                                                              Priority.normal,
                                                              owner);
            }
            if ( ! regionEndpoint.zoneDirectTargets().isEmpty()) {
                controller.nameServiceForwarder().createDirect(RecordName.from(regionEndpoint.target().name().value()),
                                                               regionEndpoint.zoneDirectTargets(),
                                                               Priority.normal,
                                                               owner);
            }
        });

        // Create global latency-based ALIAS pointing to each per-region weighted ALIAS
        controller.nameServiceForwarder().createAlias(RecordName.from(endpoint.dnsName()), latencyTargets, Priority.normal, owner);
        inactiveLatencyTargets.forEach(t -> controller.nameServiceForwarder()
                                                      .removeRecords(Record.Type.ALIAS,
                                                                     RecordName.from(endpoint.dnsName()),
                                                                     RecordData.from(t.name().value()),
                                                                     Priority.normal,
                                                                     owner));
    }

    /** Compute region endpoints and their targets from given policies */
    private Collection<RegionEndpoint> computeRegionEndpoints(Endpoint parent, List<RoutingPolicy> policies) {
        if (!parent.scope().multiDeployment()) {
            throw new IllegalStateException(parent + " has unexpected scope, got " + parent.scope());
        }
        Map<Endpoint, RegionEndpoint> endpoints = new LinkedHashMap<>();
        for (var policy : policies) {
            if (policy.dnsZone().isEmpty() && policy.canonicalName().isPresent()) continue;
            if (controller.zoneRegistry().routingMethod(policy.id().zone()) != RoutingMethod.exclusive) continue;
            var zonePolicy = db.readZoneRoutingPolicy(policy.id().zone());
            // A record with 0 weight will not receive traffic. If all records within a group have 0
            // weight, traffic is routed to all records with equal probability
            long weight = isConfiguredOut(zonePolicy, policy) ? 0 : 1;
            boolean generated = parent.generated().isPresent();
            EndpointList weightedEndpoints = controller.routing()
                                                       .endpointsOf(policy.id().deployment(),
                                                                    policy.id().cluster(),
                                                                    policy.generatedEndpoints().cluster())
                                                       .scope(Endpoint.Scope.weighted);
            if (generated) {
                weightedEndpoints = weightedEndpoints.generated();
            } else {
                weightedEndpoints = weightedEndpoints.not().generated();
            }
            if (generated && weightedEndpoints.isEmpty()) {
                // Ignore this policy. If an instance has a global endpoint, and is switching from non-generated to
                // generated endpoints we cannot update global DNS record for a deployment until it has been deployed at
                // least once (which assigns a generated endpoint).
                continue;
            }
            if (weightedEndpoints.size() != 1) {
                throw new IllegalStateException("Expected to compute exactly one region endpoint for " + policy.id() + " with parent " + parent + ", got " + weightedEndpoints);
            }
            Endpoint endpoint = weightedEndpoints.first().get();
            RegionEndpoint regionEndpoint = endpoints.computeIfAbsent(endpoint, (k) -> new RegionEndpoint(
                    new LatencyAliasTarget(DomainName.of(endpoint.dnsName()), policy.dnsZone().get(), policy.id().zone())));

            if (policy.canonicalName().isPresent()) {
                var weightedTarget = new WeightedAliasTarget(
                        policy.canonicalName().get(), policy.dnsZone().get(), policy.id().zone().value(), weight);
                regionEndpoint.add(weightedTarget);
            } else {
                var weightedTarget = new WeightedDirectTarget(
                        RecordData.from(policy.ipAddress().get()), policy.id().zone(), weight);
                regionEndpoint.add(weightedTarget);
            }
        }
        return endpoints.values();
    }


    private void updateApplicationDnsOf(RoutingPolicyList routingPolicies, DeploymentId deployment,
                                        Optional<TenantAndApplicationId> owner, @SuppressWarnings("unused") Mutex lock) {
        // In the context of single deployment (which this is) there is only one routing policy per routing ID. I.e.
        // there is no scenario where more than one deployment within an instance can be a member the same
        // application-level endpoint. However, to allow this in the future the routing table remains
        // Map<RoutingId, List<RoutingPolicy>> instead of Map<RoutingId, RoutingPolicy>.
        Map<RoutingId, List<RoutingPolicy>> routingTable = routingPolicies.asApplicationRoutingTable();
        if (routingTable.isEmpty()) return;

        Application application = controller.applications().requireApplication(routingTable.keySet().iterator().next().application());
        Map<Endpoint, Set<Target>> targetsByEndpoint = new LinkedHashMap<>();
        Map<Endpoint, Set<Target>> inactiveTargetsByEndpoint = new LinkedHashMap<>();
        for (Map.Entry<RoutingId, List<RoutingPolicy>> routeEntry : routingTable.entrySet()) {
            RoutingId routingId = routeEntry.getKey();
            EndpointList endpoints = controller.routing().readDeclaredEndpointsOf(application)
                                               .named(routingId.endpointId(), Endpoint.Scope.application);
            for (Endpoint endpoint : endpoints) {
                for (var policy : routeEntry.getValue()) {
                    for (var target : endpoint.targets()) {
                        if (!policy.appliesTo(target.deployment())) continue;
                        if (policy.dnsZone().isEmpty() && policy.canonicalName().isPresent())
                            continue; // Does not support ALIAS records
                        ZoneRoutingPolicy zonePolicy = db.readZoneRoutingPolicy(policy.id().zone());

                        Set<Target> activeTargets = targetsByEndpoint.computeIfAbsent(endpoint, (k) -> new LinkedHashSet<>());
                        Set<Target> inactiveTargets = inactiveTargetsByEndpoint.computeIfAbsent(endpoint, (k) -> new LinkedHashSet<>());
                        if (isConfiguredOut(zonePolicy, policy)) {
                            inactiveTargets.add(Target.weighted(policy, target));
                        } else {
                            activeTargets.add(Target.weighted(policy, target));
                        }
                    }
                }
            }
        }

        // Refuse removal of last target in an endpoint. We do this because removing 100% of the ALIAS records would
        // cause the application endpoint to stop resolving entirely (NXDOMAIN).
        targetsByEndpoint.forEach((endpoint, targets) -> {
            if (targets.isEmpty()) {
                throw new IllegalArgumentException("Cannot deactivate routing for " + deployment +
                                                   " as it's the last remaining active deployment in " + endpoint);
            }
        });

        // Create DNS records for active targets
        targetsByEndpoint.forEach((applicationEndpoint, targets) -> {
            // Where multiple zones are permitted, they all have the same routing policy, and nameServiceForwarder (below).
            ZoneId targetZone = applicationEndpoint.targets().iterator().next().deployment().zoneId();
            Set<AliasTarget> aliasTargets = new LinkedHashSet<>();
            Set<DirectTarget> directTargets = new LinkedHashSet<>();
            for (Target target : targets) {
                if (!target.deployment().equals(deployment)) continue; // Do not update target not matching this deployment
                if (target.aliasOrDirectTarget() instanceof AliasTarget at) {
                    aliasTargets.add(at);
                } else {
                    directTargets.add((DirectTarget) target.aliasOrDirectTarget());
                }
            }
            if (!aliasTargets.isEmpty()) {
                nameServiceForwarder(applicationEndpoint).createAlias(
                        RecordName.from(applicationEndpoint.dnsName()), aliasTargets, Priority.normal, owner);
            }
            if (!directTargets.isEmpty()) {
                nameServiceForwarder(applicationEndpoint).createDirect(
                        RecordName.from(applicationEndpoint.dnsName()), directTargets, Priority.normal, owner);
            }
        });

        // Remove DNS records for inactive targets
        inactiveTargetsByEndpoint.forEach((applicationEndpoint, targets) -> {
            targets.forEach(target -> {
                if (!target.deployment().equals(deployment)) return; // Do not update target not matching this deployment
                nameServiceForwarder(applicationEndpoint).removeRecords(target.type(),
                                                                        RecordName.from(applicationEndpoint.dnsName()),
                                                                        target.data(),
                                                                        Priority.normal,
                                                                        owner);
            });
        });
    }

    /**
     * Store routing policies for given load balancers
     *
     * @return the updated policies
     */
    private RoutingPolicyList storePoliciesOf(LoadBalancerAllocation allocation, RoutingPolicyList applicationPolicies, EndpointList generatedEndpoints, @SuppressWarnings("unused") Mutex lock) {
        Map<RoutingPolicyId, RoutingPolicy> policies = new LinkedHashMap<>(applicationPolicies.instance(allocation.deployment.applicationId()).asMap());
        for (LoadBalancer loadBalancer : allocation.loadBalancers) {
            if (loadBalancer.hostname().isEmpty() && loadBalancer.ipAddress().isEmpty()) continue;
            RoutingPolicyId policyId = new RoutingPolicyId(loadBalancer.application(), loadBalancer.cluster(), allocation.deployment.zoneId());
            RoutingPolicy existingPolicy = policies.get(policyId);
            Optional<String> dnsZone = loadBalancer.ipAddress().isPresent() ? Optional.of("ignored") : loadBalancer.dnsZone();
            List<GeneratedEndpoint> clusterGeneratedEndpoints = generatedEndpoints.cluster(loadBalancer.cluster())
                                                                                  .mapToList(e -> e.generated().get());
            clusterGeneratedEndpoints.forEach(ge -> requireNonClashing(ge, applicationPolicies.without(existingPolicy)));
            var newPolicy = new RoutingPolicy(policyId, loadBalancer.hostname(), loadBalancer.ipAddress(), dnsZone,
                                              allocation.instanceEndpointsOf(loadBalancer),
                                              allocation.applicationEndpointsOf(loadBalancer),
                                              RoutingStatus.DEFAULT,
                                              loadBalancer.isPublic(),
                                              GeneratedEndpointList.copyOf(clusterGeneratedEndpoints));
            if (existingPolicy != null) {
                newPolicy = newPolicy.with(existingPolicy.routingStatus()); // Always preserve routing status
            }
            updateZoneDnsOf(newPolicy, loadBalancer, allocation.deployment);
            policies.put(newPolicy.id(), newPolicy);
        }
        RoutingPolicyList updated = RoutingPolicyList.copyOf(policies.values());
        db.writeRoutingPolicies(allocation.deployment.applicationId(), updated.asList());
        return updated;
    }

    /** Update zone DNS record for given policy */
    private void updateZoneDnsOf(RoutingPolicy policy, LoadBalancer loadBalancer, DeploymentId deploymentId) {
        EndpointList zoneEndpoints = controller.routing().endpointsOf(deploymentId,
                                                                      policy.id().cluster(),
                                                                      policy.generatedEndpoints().cluster())
                                               .scope(Endpoint.Scope.zone);
        for (var endpoint : zoneEndpoints) {
            RecordName name = RecordName.from(endpoint.dnsName());
            Record record = policy.canonicalName().isPresent() ?
                    new Record(Record.Type.CNAME, name, RecordData.fqdn(policy.canonicalName().get().value())) :
                    new Record(Record.Type.A, name, RecordData.from(policy.ipAddress().orElseThrow()));
            nameServiceForwarder(endpoint).createRecord(record, Priority.normal, ownerOf(deploymentId));
        }
        setPrivateDns(zoneEndpoints, loadBalancer, deploymentId);
    }

    private void setPrivateDns(EndpointList endpoints, LoadBalancer loadBalancer, DeploymentId deploymentId) {
        if (loadBalancer.service().isEmpty()) return;
        // TODO(mpolden): Model one service for each endpoint (type), to allow private endpoints with tokens.
        EndpointList mtlsEndpoints = endpoints.authMethod(AuthMethod.mtls);
        if (mtlsEndpoints.isEmpty()) return;
        Endpoint endpoint = mtlsEndpoints.generated().first().orElse(mtlsEndpoints.first().get());
        if (endpoint.routingMethod() != RoutingMethod.exclusive) return; // Not supported for this routing method
        controller.serviceRegistry().vpcEndpointService()
                  .setPrivateDns(DomainName.of(endpoint.dnsName()),
                                 new ClusterId(deploymentId, endpoint.cluster()),
                                 loadBalancer.cloudAccount(),
                                 endpoint.generated().isPresent())
                  .ifPresent(challenge -> {
                      try (Mutex lock = db.lockNameServiceQueue()) {
                          controller.nameServiceForwarder().createTxt(challenge.name(), List.of(challenge.data()), Priority.high, ownerOf(deploymentId));
                          db.writeDnsChallenge(challenge);
                      }
                  });
    }

    /** Deletes all DNS challenges, and corresponding TXT records, for the given deployment. */
    public void removeDnsChallenges(DeploymentId deploymentId) {
        try (Mutex lock = db.lockNameServiceQueue()) {
            db.readDnsChallenges(deploymentId).forEach(this::removeDnsChallenge);
        }
    }

    /** Returns true iff. the given deployment has no incomplete DNS challenges, or throws (and cleans up) on errors. */
    public boolean processDnsChallenges(DeploymentId deploymentId) {
        try (Mutex lock = db.lockNameServiceQueue()) {
            List<DnsChallenge> challenges = new ArrayList<>(db.readDnsChallenges(deploymentId));
            challenges.removeIf(challenge -> challenge.state() == ChallengeState.done);
            Set<RecordName> pendingRequests = controller.curator().readNameServiceQueue().requests().stream()
                                                        .map(NameServiceRequest::name)
                                                        .collect(Collectors.toSet());
            try {
                challenges.removeIf(challenge -> {
                    if (challenge.state() == ChallengeState.pending) {
                        if (pendingRequests.contains(challenge.name())) return false;
                        challenge = challenge.withState(ChallengeState.ready);
                    }
                    ChallengeState state = controller.serviceRegistry().vpcEndpointService().process(challenge);
                    db.writeDnsChallenge(challenge.withState(state));
                    return state == ChallengeState.done;
                });
                return challenges.isEmpty();
            }
            catch (RuntimeException e) {
                challenges.forEach(this::removeDnsChallenge);
                throw e;
            }
        }
    }

    private void removeDnsChallenge(DnsChallenge challenge) {
        controller.nameServiceForwarder().removeRecords(Record.Type.TXT, challenge.name(), Priority.normal, ownerOf(challenge.clusterId().deploymentId()));
        db.deleteDnsChallenge(challenge.clusterId());
    }

    /**
     * Remove policies and zone DNS records unreferenced by given load balancers
     *
     * @return the updated policies
     */
    private RoutingPolicyList removePoliciesUnreferencedBy(LoadBalancerAllocation allocation, RoutingPolicyList instancePolicies, @SuppressWarnings("unused") Mutex lock) {
        Map<RoutingPolicyId, RoutingPolicy> newPolicies = new LinkedHashMap<>(instancePolicies.asMap());
        Set<RoutingPolicyId> activeIds = allocation.asPolicyIds();
        RoutingPolicyList removable = instancePolicies.deployment(allocation.deployment)
                                                      .not().matching(policy -> activeIds.contains(policy.id()));
        for (var policy : removable) {
            EndpointList zoneEndpoints = controller.routing().endpointsOf(allocation.deployment,
                                                                          policy.id().cluster(),
                                                                          policy.generatedEndpoints().cluster())
                                                   .scope(Endpoint.Scope.zone);
            for (var endpoint : zoneEndpoints) {
                Record.Type type = policy.canonicalName().isPresent() ? Record.Type.CNAME : Record.Type.A;
                nameServiceForwarder(endpoint).removeRecords(type,
                                                             RecordName.from(endpoint.dnsName()),
                                                             Priority.normal,
                                                             ownerOf(allocation));
            }
            newPolicies.remove(policy.id());
        }
        RoutingPolicyList updated = RoutingPolicyList.copyOf(newPolicies.values());
        db.writeRoutingPolicies(allocation.deployment.applicationId(), updated.asList());
        return updated;
    }

    /** Remove unreferenced instance endpoints from DNS */
    private void removeGlobalDnsUnreferencedBy(LoadBalancerAllocation allocation, RoutingPolicyList deploymentPolicies, @SuppressWarnings("unused") Mutex lock) {
        Map<RoutingId, List<RoutingPolicy>> routingTable = deploymentPolicies.asInstanceRoutingTable();
        Set<RoutingId> removalCandidates = new HashSet<>(routingTable.keySet());
        Set<RoutingId> activeRoutingIds = instanceRoutingIds(allocation);
        removalCandidates.removeAll(activeRoutingIds);
        for (var id : removalCandidates) {
            List<RoutingPolicy> policies = routingTable.get(id);
            Map<ClusterSpec.Id, List<RoutingPolicy>> policyByCluster = policies.stream().collect(Collectors.groupingBy(p -> p.id().cluster()));
            Set<Endpoint> endpoints = new LinkedHashSet<>();
            policyByCluster.forEach((cluster, clusterPolicies) -> {
                List<DeploymentId> deployments = clusterPolicies.stream().map(p -> p.id().deployment()).toList();
                GeneratedEndpointList generated = declaredGeneratedEndpoints(id.endpointId(), clusterPolicies);
                endpoints.addAll(controller.routing().declaredEndpointsOf(id, cluster, deployments, generated)
                                           .not().requiresRotation()
                                           .named(id.endpointId(), Endpoint.Scope.global).asList());
            });
            // This removes all ALIAS records having this DNS name. There is no attempt to delete only the entry for the
            // affected zone. Instead, the correct set of records is (re)created by updateGlobalDnsOf
            for (var endpoint : endpoints) {
                for (var regionEndpoint : computeRegionEndpoints(endpoint, deploymentPolicies.asList())) {
                    Record.Type type = regionEndpoint.zoneDirectTargets().isEmpty() ? Record.Type.ALIAS : Record.Type.DIRECT;
                    controller.nameServiceForwarder().removeRecords(type,
                                                                    RecordName.from(regionEndpoint.target().name().value()),
                                                                    Priority.normal,
                                                                    ownerOf(allocation));
                }
                nameServiceForwarder(endpoint).removeRecords(Record.Type.ALIAS, RecordName.from(endpoint.dnsName()),
                                                             Priority.normal,
                                                             ownerOf(allocation));
            }
        }
    }

    /** Remove unreferenced application endpoints in given allocation from DNS */
    private void removeApplicationDnsUnreferencedBy(LoadBalancerAllocation allocation, RoutingPolicyList deploymentPolicies, @SuppressWarnings("unused") Mutex lock) {
        Map<RoutingId, List<RoutingPolicy>> routingTable = deploymentPolicies.asApplicationRoutingTable();
        Set<RoutingId> removalCandidates = new HashSet<>(routingTable.keySet());
        Set<RoutingId> activeRoutingIds = applicationRoutingIds(allocation);
        removalCandidates.removeAll(activeRoutingIds);
        for (var id : removalCandidates) {
            TenantAndApplicationId application = TenantAndApplicationId.from(id.instance());
            List<RoutingPolicy> policies = routingTable.get(id);
            Map<ClusterSpec.Id, List<RoutingPolicy>> policyByCluster = policies.stream().collect(Collectors.groupingBy(p -> p.id().cluster()));
            Set<Endpoint> endpoints = new LinkedHashSet<>();
            policyByCluster.forEach((cluster, clusterPolicies) -> {
                // Weights are not available in this context, but they're not used for anything when removing records
                Map<DeploymentId, Integer> deployments = clusterPolicies.stream()
                                                                 .map(p -> p.id().deployment())
                                                                 .collect(Collectors.toMap(Function.identity(), (ignored) -> 1));
                GeneratedEndpointList generated = declaredGeneratedEndpoints(id.endpointId(), clusterPolicies);
                endpoints.addAll(controller.routing().declaredEndpointsOf(application, id.endpointId(), cluster,
                                                                          deployments, generated).asList());
            });
            for (var policy : policies) {
                if (!policy.appliesTo(allocation.deployment)) continue;
                for (Endpoint endpoint : endpoints) {
                    NameServiceForwarder forwarder = nameServiceForwarder(endpoint);
                    if (policy.canonicalName().isPresent()) {
                        forwarder.removeRecords(Record.Type.ALIAS,
                                                RecordName.from(endpoint.dnsName()),
                                                RecordData.fqdn(policy.canonicalName().get().value()),
                                                Priority.normal,
                                                ownerOf(allocation));
                    } else {
                        forwarder.removeRecords(Record.Type.DIRECT,
                                                RecordName.from(endpoint.dnsName()),
                                                RecordData.from(policy.ipAddress().get()),
                                                Priority.normal,
                                                ownerOf(allocation));
                    }
                }
            }
        }
    }

    private Set<RoutingId> instanceRoutingIds(LoadBalancerAllocation allocation) {
        return routingIdsFrom(allocation, false);
    }

    private Set<RoutingId> applicationRoutingIds(LoadBalancerAllocation allocation) {
        return routingIdsFrom(allocation, true);
    }

    private static GeneratedEndpointList declaredGeneratedEndpoints(EndpointId endpoint, List<RoutingPolicy> clusterPolicies) {
        return GeneratedEndpointList.copyOf(clusterPolicies.stream()
                                                           .flatMap(p -> p.generatedEndpoints().declared(endpoint).asList().stream())
                                                           .distinct()
                                                           .toList());
    }

    /** Compute routing IDs from given load balancers */
    private static Set<RoutingId> routingIdsFrom(LoadBalancerAllocation allocation, boolean applicationLevel) {
        Set<RoutingId> routingIds = new LinkedHashSet<>();
        for (var loadBalancer : allocation.loadBalancers) {
            Set<EndpointId> endpoints = applicationLevel
                    ? allocation.applicationEndpointsOf(loadBalancer)
                    : allocation.instanceEndpointsOf(loadBalancer);
            for (var endpointId : endpoints) {
                routingIds.add(RoutingId.of(loadBalancer.application(), endpointId));
            }
        }
        return Collections.unmodifiableSet(routingIds);
    }

    /** Returns whether the endpoints of given policy are configured {@link RoutingStatus.Value#out} */
    private static boolean isConfiguredOut(ZoneRoutingPolicy zonePolicy, RoutingPolicy policy) {
        // A deployment can be configured out from endpoints at any of the following levels:
        // - zone level (ZoneRoutingPolicy)
        // - deployment level (RoutingPolicy)
        return zonePolicy.routingStatus().value() == RoutingStatus.Value.out ||
               policy.routingStatus().value() == RoutingStatus.Value.out;
    }

    /** Represents records for a region-wide endpoint */
    private static class RegionEndpoint {

        private final LatencyAliasTarget target;
        private final Set<WeightedAliasTarget> zoneAliasTargets = new LinkedHashSet<>();
        private final Set<WeightedDirectTarget> zoneDirectTargets = new LinkedHashSet<>();

        public RegionEndpoint(LatencyAliasTarget target) {
            this.target = Objects.requireNonNull(target);
        }

        public LatencyAliasTarget target() { return target; }
        public Set<AliasTarget> zoneAliasTargets() { return Collections.unmodifiableSet(zoneAliasTargets); }
        public Set<DirectTarget> zoneDirectTargets() { return Collections.unmodifiableSet(zoneDirectTargets); }

        public void add(WeightedAliasTarget target) { zoneAliasTargets.add(target); }
        public void add(WeightedDirectTarget target) { zoneDirectTargets.add(target); }

        public boolean active() {
            return zoneAliasTargets.stream().anyMatch(target -> target.weight() > 0) ||
                   zoneDirectTargets.stream().anyMatch(target -> target.weight() > 0);
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            RegionEndpoint that = (RegionEndpoint) o;
            return target.name().equals(that.target.name());
        }

        @Override
        public int hashCode() {
            return Objects.hash(target.name());
        }

    }

    /** Active load balancers allocated to a deployment */
    record LoadBalancerAllocation(DeploymentId deployment,
                                  DeploymentSpec deploymentSpec,
                                  List<LoadBalancer> loadBalancers) {

        public LoadBalancerAllocation(DeploymentId deployment,
                                      DeploymentSpec deploymentSpec,
                                      List<LoadBalancer> loadBalancers) {
            this.deployment = deployment;
            this.loadBalancers = loadBalancers.stream().filter(LoadBalancerAllocation::isActive).toList();
            this.deploymentSpec = deploymentSpec;
        }

        private static boolean isActive(LoadBalancer loadBalancer) {
            return switch (loadBalancer.state()) {
                // Count reserved as active as we want to do DNS updates as early as possible
                case reserved, active -> true;
                default -> false;
            };
        }

        /** Returns the policy IDs of the load balancers contained in this */
        private Set<RoutingPolicyId> asPolicyIds() {
            return loadBalancers.stream()
                                .map(lb -> new RoutingPolicyId(lb.application(),
                                                               lb.cluster(),
                                                               deployment.zoneId()))
                                .collect(Collectors.toUnmodifiableSet());
        }

        /** Returns all instance endpoint IDs served by given load balancer */
        private Set<EndpointId> instanceEndpointsOf(LoadBalancer loadBalancer) {
            if (!deployment.zoneId().environment().isProduction()) { // Only production deployments have configurable endpoints
                return Set.of();
            }
            var instanceSpec = deploymentSpec.instance(loadBalancer.application().instance());
            if (instanceSpec.isEmpty()) {
                return Set.of();
            }
            return instanceSpec.get().endpoints().stream()
                               .filter(endpoint -> endpoint.containerId().equals(loadBalancer.cluster().value()))
                               .filter(endpoint -> endpoint.regions().contains(deployment.zoneId().region()))
                               .map(com.yahoo.config.application.api.Endpoint::endpointId)
                               .map(EndpointId::of)
                               .collect(Collectors.toUnmodifiableSet());
        }

        /** Returns all application endpoint IDs served by given load balancer */
        private Set<EndpointId> applicationEndpointsOf(LoadBalancer loadBalancer) {
            if (!deployment.zoneId().environment().isProduction()) { // Only production deployments have configurable endpoints
                return Set.of();
            }
            return deploymentSpec.endpoints().stream()
                                 .filter(endpoint -> endpoint.containerId().equals(loadBalancer.cluster().value()))
                                 .filter(endpoint -> endpoint.targets().stream()
                                                             .anyMatch(target -> target.region().equals(deployment.zoneId().region()) &&
                                                                                 target.instance().equals(deployment.applicationId().instance())))
                                 .map(com.yahoo.config.application.api.Endpoint::endpointId)
                                 .map(EndpointId::of)
                                 .collect(Collectors.toUnmodifiableSet());
        }

    }

    /** Returns the name updater to use for given endpoint */
    private NameServiceForwarder nameServiceForwarder(Endpoint endpoint) {
        return switch (endpoint.routingMethod()) {
            case exclusive -> controller.nameServiceForwarder();
            case sharedLayer4 -> endpoint.generated().isPresent() ? controller.nameServiceForwarder() : new NameServiceDiscarder(controller.curator());
        };
    }

    /** Denotes record data (record rhs) of either an ALIAS or a DIRECT target */
    private record Target(Record.Type type, RecordData data, DeploymentId deployment, Object aliasOrDirectTarget) {
        static Target weighted(RoutingPolicy policy, Endpoint.Target endpointTarget) {
            if (policy.ipAddress().isPresent()) {
                var wt = new WeightedDirectTarget(RecordData.from(policy.ipAddress().get()),
                        endpointTarget.deployment().zoneId(), endpointTarget.weight());
                return new Target(Record.Type.DIRECT, wt.recordData(), endpointTarget.deployment(), wt);
            }
            var wt = new WeightedAliasTarget(policy.canonicalName().get(), policy.dnsZone().get(),
                    endpointTarget.deployment().zoneId().value(), endpointTarget.weight());
            return new Target(Record.Type.ALIAS, RecordData.fqdn(wt.name().value()), endpointTarget.deployment(), wt);
        }
    }

    /** A {@link NameServiceForwarder} that does nothing. Used in zones where no explicit DNS updates are needed */
    private static class NameServiceDiscarder extends NameServiceForwarder {

        public NameServiceDiscarder(CuratorDb db) {
            super(db);
        }

        @Override
        protected void forward(NameServiceRequest request, Priority priority) {
            // Ignored
        }
    }

    private static Optional<TenantAndApplicationId> ownerOf(DeploymentId deploymentId) {
        return Optional.of(TenantAndApplicationId.from(deploymentId.applicationId()));
    }

    private static Optional<TenantAndApplicationId> ownerOf(LoadBalancerAllocation allocation) {
        return ownerOf(allocation.deployment);
    }

    private static void requireNonClashing(GeneratedEndpoint generatedEndpoint, RoutingPolicyList applicationPolicies) {
        for (var policy : applicationPolicies) {
            for (var other : policy.generatedEndpoints()) {
                if (other.clusterPart().equals(generatedEndpoint.clusterPart()) && !other.endpoint().equals(generatedEndpoint.endpoint())) {
                    throw new IllegalStateException(generatedEndpoint + " clashes with " + other + " in " + policy.id());
                }
            }
        }
    }

}