/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.  The ASF licenses this file to you under the Apache License, Version
 * 2.0 (the "License"); you may not use this file except in compliance with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions
 * and limitations under the License.
 */

package org.apache.storm.scheduler;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import org.apache.commons.lang.Validate;
import org.apache.storm.DaemonConfig;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// for each isolated topology:
//   compute even distribution of executors -> workers on the number of workers specified for the topology
//   compute distribution of workers to machines
// determine host -> list of [slot, topology id, executors]
// iterate through hosts and: a machine is good if:
//   1. only running workers from one isolated topology
//   2. all workers running on it match one of the distributions of executors for that topology
//   3. matches one of the # of workers
// blacklist the good hosts and remove those workers from the list of need to be assigned workers
// otherwise unassign all other workers for isolated topologies if assigned
public class IsolationScheduler implements IScheduler {
    private static final Logger LOG = LoggerFactory.getLogger(IsolationScheduler.class);

    private Map<String, Number> isoMachines;

    @Override
    public void prepare(Map<String, Object> conf, StormMetricsRegistry metricsRegistry) {
        this.isoMachines = (Map<String, Number>) conf.get(DaemonConfig.ISOLATION_SCHEDULER_MACHINES);
        Validate.notEmpty(isoMachines);
    }

    @Override
    public Map<String, Map<String, Double>> config() {
        return Collections.emptyMap();
    }

    // get host -> all assignable worker slots for non-blacklisted machines (assigned or not assigned)
    // will then have a list of machines that need to be assigned (machine -> [topology, list of list of executors])
    // match each spec to a machine (who has the right number of workers), free everything else on that machine and assign those slots
    // (do one topology at a time)
    // blacklist all machines who had production slots defined
    // log isolated topologies who weren't able to get enough slots / machines
    // run default scheduler on isolated topologies that didn't have enough slots + non-isolated topologies on remaining machines
    // set blacklist to what it was initially
    @Override
    public void schedule(Topologies topologies, Cluster cluster) {
        List<TopologyDetails> isoTopologies = isolatedTopologies(topologies.getTopologies());
        Set<String> isoIds = extractTopologyIds(isoTopologies);
        Map<String, Set<Set<ExecutorDetails>>> topologyWorkerSpecs = topologyWorkerSpecs(isoTopologies);
        Map<String, Map<Integer, Integer>> topologyMachineDistributions = topologyMachineDistributions(isoTopologies);
        Map<String, List<AssignmentInfo>> hostAssignments = hostAssignments(cluster);

        for (Map.Entry<String, List<AssignmentInfo>> entry : hostAssignments.entrySet()) {
            List<AssignmentInfo> assignments = entry.getValue();
            String topologyId = assignments.get(0).getTopologyId();
            Map<Integer, Integer> distribution = topologyMachineDistributions.get(topologyId);
            Set<Set<ExecutorDetails>> workerSpecs = topologyWorkerSpecs.get(topologyId);
            int numWorkers = assignments.size();

            if (isoIds.contains(topologyId)
                && checkAssignmentTopology(assignments, topologyId)
                && distribution.containsKey(numWorkers)
                && checkAssignmentWorkerSpecs(assignments, workerSpecs)) {
                decrementDistribution(distribution, numWorkers);
                for (AssignmentInfo ass : assignments) {
                    workerSpecs.remove(ass.getExecutors());
                }
                cluster.blacklistHost(entry.getKey());
            } else {
                for (AssignmentInfo ass : assignments) {
                    if (isoIds.contains(ass.getTopologyId())) {
                        cluster.freeSlot(ass.getWorkerSlot());
                    }
                }
            }
        }

        Map<String, Set<WorkerSlot>> hostUsedSlots = hostToUsedSlots(cluster);
        LinkedList<HostAssignableSlots> hss = hostAssignableSlots(cluster);
        for (Map.Entry<String, Set<Set<ExecutorDetails>>> entry : topologyWorkerSpecs.entrySet()) {
            String topologyId = entry.getKey();
            Set<Set<ExecutorDetails>> executorSet = entry.getValue();
            List<Integer> workerNum = distributionToSortedAmounts(topologyMachineDistributions.get(topologyId));
            for (Integer num : workerNum) {
                HostAssignableSlots hostSlots = hss.peek();
                List<WorkerSlot> slot = hostSlots != null ? hostSlots.getWorkerSlots() : null;

                if (slot != null && slot.size() >= num) {
                    hss.poll();
                    cluster.freeSlots(hostUsedSlots.get(hostSlots.getHostName()));
                    for (WorkerSlot tmpSlot : slot.subList(0, num)) {
                        Set<ExecutorDetails> executor = removeElemFromExecutorsSet(executorSet);
                        cluster.assign(tmpSlot, topologyId, executor);
                    }
                    cluster.blacklistHost(hostSlots.getHostName());
                }
            }
        }

        List<String> failedTopologyIds = extractFailedTopologyIds(topologyWorkerSpecs);
        if (failedTopologyIds.size() > 0) {
            LOG.warn("Unable to isolate topologies " + failedTopologyIds
                     + ". No machine had enough worker slots to run the remaining workers for these topologies. "
                     + "Clearing all other resources and will wait for enough resources for "
                     + "isolated topologies before allocating any other resources.");
            // clear workers off all hosts that are not blacklisted
            Map<String, Set<WorkerSlot>> usedSlots = hostToUsedSlots(cluster);
            Set<Map.Entry<String, Set<WorkerSlot>>> entries = usedSlots.entrySet();
            for (Map.Entry<String, Set<WorkerSlot>> entry : entries) {
                if (!cluster.isBlacklistedHost(entry.getKey())) {
                    cluster.freeSlots(entry.getValue());
                }
            }
        } else {
            // run default scheduler on non-isolated topologies
            Set<String> allocatedTopologies = allocatedTopologies(topologyWorkerSpecs);
            Topologies leftOverTopologies = leftoverTopologies(topologies, allocatedTopologies);
            DefaultScheduler.defaultSchedule(leftOverTopologies, cluster);
        }
        Set<String> origBlacklist = cluster.getBlacklistedHosts();
        cluster.setBlacklistedHosts(origBlacklist);
    }

    private Set<ExecutorDetails> removeElemFromExecutorsSet(Set<Set<ExecutorDetails>> executorsSets) {
        Set<ExecutorDetails> elem = executorsSets.iterator().next();
        executorsSets.remove(elem);
        return elem;
    }

    private List<TopologyDetails> isolatedTopologies(Collection<TopologyDetails> topologies) {
        Set<String> topologyNames = isoMachines.keySet();
        List<TopologyDetails> isoTopologies = new ArrayList<TopologyDetails>();
        for (TopologyDetails topo : topologies) {
            if (topologyNames.contains(topo.getName())) {
                isoTopologies.add(topo);
            }
        }
        return isoTopologies;
    }

    private Set<String> extractTopologyIds(List<TopologyDetails> topologies) {
        Set<String> ids = new HashSet<String>();
        if (topologies != null && topologies.size() > 0) {
            for (TopologyDetails topology : topologies) {
                ids.add(topology.getId());
            }
        }
        return ids;
    }

    private List<String> extractFailedTopologyIds(Map<String, Set<Set<ExecutorDetails>>> isoTopologyWorkerSpecs) {
        List<String> failedTopologyIds = new ArrayList<String>();
        for (Map.Entry<String, Set<Set<ExecutorDetails>>> topoWorkerSpecsEntry : isoTopologyWorkerSpecs.entrySet()) {
            Set<Set<ExecutorDetails>> workerSpecs = topoWorkerSpecsEntry.getValue();
            if (workerSpecs != null && !workerSpecs.isEmpty()) {
                failedTopologyIds.add(topoWorkerSpecsEntry.getKey());
            }
        }
        return failedTopologyIds;
    }

    // map from topology id -> set of sets of executors
    private Map<String, Set<Set<ExecutorDetails>>> topologyWorkerSpecs(List<TopologyDetails> topologies) {
        Map<String, Set<Set<ExecutorDetails>>> workerSpecs = new HashMap<String, Set<Set<ExecutorDetails>>>();
        for (TopologyDetails topology : topologies) {
            workerSpecs.put(topology.getId(), computeWorkerSpecs(topology));
        }
        return workerSpecs;
    }

    private Map<String, List<AssignmentInfo>> hostAssignments(Cluster cluster) {
        Collection<SchedulerAssignment> assignmentValues = cluster.getAssignments().values();
        Map<String, List<AssignmentInfo>> hostAssignments = new HashMap<String, List<AssignmentInfo>>();

        for (SchedulerAssignment sa : assignmentValues) {
            Map<WorkerSlot, List<ExecutorDetails>> slotExecutors = Utils.reverseMap(sa.getExecutorToSlot());
            Set<Map.Entry<WorkerSlot, List<ExecutorDetails>>> entries = slotExecutors.entrySet();
            for (Map.Entry<WorkerSlot, List<ExecutorDetails>> entry : entries) {
                WorkerSlot slot = entry.getKey();
                List<ExecutorDetails> executors = entry.getValue();

                String host = cluster.getHost(slot.getNodeId());
                AssignmentInfo ass = new AssignmentInfo(slot, sa.getTopologyId(), new HashSet<ExecutorDetails>(executors));
                List<AssignmentInfo> executorList = hostAssignments.get(host);
                if (executorList == null) {
                    executorList = new ArrayList<AssignmentInfo>();
                    hostAssignments.put(host, executorList);
                }
                executorList.add(ass);
            }
        }
        return hostAssignments;
    }

    private Set<Set<ExecutorDetails>> computeWorkerSpecs(TopologyDetails topology) {
        Map<String, List<ExecutorDetails>> compExecutors = Utils.reverseMap(topology.getExecutorToComponent());

        List<ExecutorDetails> allExecutors = new ArrayList<ExecutorDetails>();
        Collection<List<ExecutorDetails>> values = compExecutors.values();
        for (List<ExecutorDetails> value : values) {
            allExecutors.addAll(value);
        }

        int numWorkers = topology.getNumWorkers();
        int bucketIndex = 0;
        Map<Integer, Set<ExecutorDetails>> bucketExecutors = new HashMap<Integer, Set<ExecutorDetails>>(numWorkers);
        for (ExecutorDetails executor : allExecutors) {
            Set<ExecutorDetails> executors = bucketExecutors.get(bucketIndex);
            if (executors == null) {
                executors = new HashSet<ExecutorDetails>();
                bucketExecutors.put(bucketIndex, executors);
            }
            executors.add(executor);
            bucketIndex = (bucketIndex + 1) % numWorkers;
        }

        return new HashSet<Set<ExecutorDetails>>(bucketExecutors.values());
    }

    private Map<String, Map<Integer, Integer>> topologyMachineDistributions(List<TopologyDetails> isoTopologies) {
        Map<String, Map<Integer, Integer>> machineDistributions = new HashMap<String, Map<Integer, Integer>>();
        for (TopologyDetails topology : isoTopologies) {
            machineDistributions.put(topology.getId(), machineDistribution(topology));
        }
        return machineDistributions;
    }

    private Map<Integer, Integer> machineDistribution(TopologyDetails topology) {
        int machineNum = isoMachines.get(topology.getName()).intValue();
        int workerNum = topology.getNumWorkers();
        TreeMap<Integer, Integer> distribution = Utils.integerDivided(workerNum, machineNum);

        if (distribution.containsKey(0)) {
            distribution.remove(0);
        }
        return distribution;
    }

    private boolean checkAssignmentTopology(List<AssignmentInfo> assignments, String topologyId) {
        for (AssignmentInfo ass : assignments) {
            if (!topologyId.equals(ass.getTopologyId())) {
                return false;
            }
        }
        return true;
    }

    private boolean checkAssignmentWorkerSpecs(List<AssignmentInfo> assigments, Set<Set<ExecutorDetails>> workerSpecs) {
        for (AssignmentInfo ass : assigments) {
            if (!workerSpecs.contains(ass.getExecutors())) {
                return false;
            }
        }
        return true;
    }

    private void decrementDistribution(Map<Integer, Integer> distribution, int value) {
        Integer distValue = distribution.get(value);
        if (distValue != null) {
            int newValue = distValue - 1;
            if (newValue == 0) {
                distribution.remove(value);
            } else {
                distribution.put(value, newValue);
            }
        }
    }

    private Map<String, Set<WorkerSlot>> hostToUsedSlots(Cluster cluster) {
        Collection<WorkerSlot> usedSlots = cluster.getUsedSlots();
        Map<String, Set<WorkerSlot>> hostUsedSlots = new HashMap<String, Set<WorkerSlot>>();
        for (WorkerSlot slot : usedSlots) {
            String host = cluster.getHost(slot.getNodeId());
            Set<WorkerSlot> slots = hostUsedSlots.get(host);
            if (slots == null) {
                slots = new HashSet<WorkerSlot>();
                hostUsedSlots.put(host, slots);
            }
            slots.add(slot);
        }
        return hostUsedSlots;
    }

    // returns list of list of slots, reverse sorted by number of slots
    private LinkedList<HostAssignableSlots> hostAssignableSlots(Cluster cluster) {
        List<WorkerSlot> assignableSlots = cluster.getAssignableSlots();
        Map<String, List<WorkerSlot>> hostAssignableSlots = new HashMap<String, List<WorkerSlot>>();
        for (WorkerSlot slot : assignableSlots) {
            String host = cluster.getHost(slot.getNodeId());
            List<WorkerSlot> slots = hostAssignableSlots.get(host);
            if (slots == null) {
                slots = new ArrayList<WorkerSlot>();
                hostAssignableSlots.put(host, slots);
            }
            slots.add(slot);
        }
        List<HostAssignableSlots> sortHostAssignSlots = new ArrayList<HostAssignableSlots>();
        for (Map.Entry<String, List<WorkerSlot>> entry : hostAssignableSlots.entrySet()) {
            sortHostAssignSlots.add(new HostAssignableSlots(entry.getKey(), entry.getValue()));
        }
        Collections.sort(sortHostAssignSlots, new Comparator<HostAssignableSlots>() {
            @Override
            public int compare(HostAssignableSlots o1, HostAssignableSlots o2) {
                return o2.getWorkerSlots().size() - o1.getWorkerSlots().size();
            }
        });
        Collections.shuffle(sortHostAssignSlots);

        return new LinkedList<HostAssignableSlots>(sortHostAssignSlots);
    }

    private List<Integer> distributionToSortedAmounts(Map<Integer, Integer> distributions) {
        List<Integer> sorts = new ArrayList<Integer>();
        for (Map.Entry<Integer, Integer> entry : distributions.entrySet()) {
            int workers = entry.getKey();
            int machines = entry.getValue();
            for (int i = 0; i < machines; i++) {
                sorts.add(workers);
            }
        }
        Collections.sort(sorts, new Comparator<Integer>() {
            @Override
            public int compare(Integer o1, Integer o2) {
                return o2.intValue() - o1.intValue();
            }
        });

        return sorts;
    }

    private Set<String> allocatedTopologies(Map<String, Set<Set<ExecutorDetails>>> topologyToWorkerSpecs) {
        Set<String> allocatedTopologies = new HashSet<String>();
        Set<Map.Entry<String, Set<Set<ExecutorDetails>>>> entries = topologyToWorkerSpecs.entrySet();
        for (Map.Entry<String, Set<Set<ExecutorDetails>>> entry : entries) {
            if (entry.getValue().isEmpty()) {
                allocatedTopologies.add(entry.getKey());
            }
        }
        return allocatedTopologies;
    }

    private Topologies leftoverTopologies(Topologies topologies, Set<String> filterIds) {
        Collection<TopologyDetails> topos = topologies.getTopologies();
        Map<String, TopologyDetails> leftoverTopologies = new HashMap<String, TopologyDetails>();
        for (TopologyDetails topo : topos) {
            String id = topo.getId();
            if (!filterIds.contains(id)) {
                leftoverTopologies.put(id, topo);
            }
        }
        return new Topologies(leftoverTopologies);
    }

    class AssignmentInfo {
        private WorkerSlot workerSlot;
        private String topologyId;
        private Set<ExecutorDetails> executors;

        AssignmentInfo(WorkerSlot workerSlot, String topologyId, Set<ExecutorDetails> executors) {
            this.workerSlot = workerSlot;
            this.topologyId = topologyId;
            this.executors = executors;
        }

        public WorkerSlot getWorkerSlot() {
            return workerSlot;
        }

        public String getTopologyId() {
            return topologyId;
        }

        public Set<ExecutorDetails> getExecutors() {
            return executors;
        }

    }

    class HostAssignableSlots {
        private String hostName;
        private List<WorkerSlot> workerSlots;

        HostAssignableSlots(String hostName, List<WorkerSlot> workerSlots) {
            this.hostName = hostName;
            this.workerSlots = workerSlots;
        }

        public String getHostName() {
            return hostName;
        }

        public List<WorkerSlot> getWorkerSlots() {
            return workerSlots;
        }

    }
}