/*-
 *
 *  This file is part of Oracle NoSQL Database
 *  Copyright (C) 2011, 2014 Oracle and/or its affiliates.  All rights reserved.
 *
 *  Oracle NoSQL Database is free software: you can redistribute it and/or
 *  modify it under the terms of the GNU Affero General Public License
 *  as published by the Free Software Foundation, version 3.
 *
 *  Oracle NoSQL Database is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 *  Affero General Public License for more details.
 *
 *  You should have received a copy of the GNU Affero General Public
 *  License in the LICENSE file along with Oracle NoSQL Database.  If not,
 *  see <http://www.gnu.org/licenses/>.
 *
 *  An active Oracle commercial licensing agreement for this product
 *  supercedes this license.
 *
 *  For more information please contact:
 *
 *  Vice President Legal, Development
 *  Oracle America, Inc.
 *  5OP-10
 *  500 Oracle Parkway
 *  Redwood Shores, CA 94065
 *
 *  or
 *
 *  berkeleydb-info_us@oracle.com
 *
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  [This line intentionally left blank.]
 *  EOF
 *
 */

package oracle.kv.impl.admin.topo;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicInteger;

import oracle.kv.impl.admin.param.Parameters;
import oracle.kv.impl.admin.param.RepNodeParams;
import oracle.kv.impl.topo.Partition;
import oracle.kv.impl.topo.PartitionId;
import oracle.kv.impl.topo.RepGroupId;
import oracle.kv.impl.topo.RepNodeId;
import oracle.kv.impl.topo.StorageNodeId;
import oracle.kv.impl.topo.Topology;

/**
 * Compare and categorize the differences between two topologies. Used both as
 * an initial step when constructing a deployment plan and for supplying an
 * user-viewable preview of the work required to transform from one topology
 * to another.
 */

public class TopologyDiff {

    private static final String CURRENT = "current deployed topology";
    public static final String NO_CHANGE = "No differences in topologies.";

    private final Topology source;
    private final String sourceName;
    private final TopologyCandidate candidate;
    private final Parameters params;

    /*
     * Brand new shards, listed in ordinal value. The RepGroupIds are from the
     * source topology. That is, newShard.get(0) is the repGroupId for the
     * first new shard, from the source topo. When the topology is deployed,
     * the shards may not be created in precisely ordinal order, if there is
     * some parallelism, and the actual repGroupId that is allocated may be
     * different.
     */
    private final List<RepGroupId> newShards;

    /*
     * A record of all changes to all shards, such as relocated or newly
     * created RNS, or relocated partitions. At the time when the diff is
     * generated, the repGroupId of the shard in the source and candidate
     * topologies are obviously the same, but when the tasks for the topology
     * are generated by the TopoTaskGenerator, the resulting shards might have
     * ids that are different, because component id generation it controlled
     * transparently by the Topology.
     */
    private final Map<RepGroupId, ShardChange> changedShards;

    private final Set<RepNodeId> newRNs;

    /* ===== Topology components that are now unused. ===== */

    /*
     * Note that in R2, contraction is not supported, and RNs and shards
     * should not be removed.
     */

    /* Shards that are in the source but not the candidate. */
    private final Set<RepGroupId> removedShards;

    /* ===== For reporting and validation ===== */

    /*
     * Shards that are sources for partition migration. Currently used for
     * stats, will be used for optimizing task ordering in the future.
     */
    private final Set<RepGroupId> pMigrationSources;

    /*
     * Shards that are destinations for partition migration. Currently used for
     * stats, will be used for optimizing task ordering in the future. When
     * displaying this, take care to account for the fact that the destination
     * shard id will not truly have been created yet; the RepGroupId used as
     * a destination id is a placeholder until the shard has truly been created
     * in the topology.
     */
    private final Set<RepGroupId> pMigrationDestinations;

    private int numCreatedPartitions;
    private int numPartitionMigrations;
    private int numRelocatedRNs;

    /** Mount point assignments created by the topology building phase. */
    private Map<RepNodeId, String> mountPointAssignments;

    /**
     * Evaluate the differences between the source and destination, or
     * candidate.
     *
     * @param source the original topology
     * @param sourceName the candidate name of the original topology. If null,
     * the default name of CURRENT will be used.
     * @param candidate the destination topology.
     */
    public TopologyDiff(Topology source, String sourceName,
                        TopologyCandidate candidate, Parameters params) {
        this(source, sourceName, candidate, params, true /*validate*/);
    }

    /**
     * Evaluate the differences between the source and destination, or
     * candidate. Optionally validate the transition.
     *
     * @param source the original topology
     * @param sourceName the candidate name of the original topology. If null,
     * the default name of CURRENT will be used.
     * @param candidate the destination topology.
     */
    public TopologyDiff(Topology source, String sourceName,
                        TopologyCandidate candidate, Parameters params,
                        boolean validate) {
        this.source = source;
        if (sourceName == null) {
            this.sourceName = CURRENT;
        } else {
            this.sourceName = sourceName;
        }
        this.candidate = candidate;
        this.params = params;

        /*
         * Sort the change set info by shard name, so the preview/display is
         * nice, and so that the generated topology transformation tasks are
         * deterministic and more intuitive to the user.
         */
        changedShards = new TreeMap<RepGroupId,ShardChange>
            (new Comparator<RepGroupId>() {
                @Override
                public
                int compare(RepGroupId r1, RepGroupId r2) {
                    return r1.getGroupId() - r2.getGroupId();
                }
            });

        numCreatedPartitions = 0;

        newShards = new ArrayList<RepGroupId>();
        removedShards = new HashSet<RepGroupId>();
        newRNs = new HashSet<RepNodeId>();

        pMigrationSources = new HashSet<RepGroupId>();
        pMigrationDestinations = new HashSet<RepGroupId>();

        doDiff(validate);
    }

    /*
     * Methods to interrogate the diff. Should there be more, or are these
     * unnecessary?
     */
    public int getNumCreatedPartitions() {
      return numCreatedPartitions;
    }

    /**
     * Return the number of new shards to create.
     */
    public int getNumNewShards() {
        return newShards.size();
    }

    /**
     * Return the number of new RNs to create.
     */
    public int getNumNewRNs() {
        return newRNs.size();
    }

    /**
     * Return the number of RNs to relocate.
     */
    public int getNumRelocatedRNs() {
        return numRelocatedRNs;
    }

    /**
     * Return a listing of the work to be done. Format is
     * Summary:
     * Create N shards
     * Create N RNs
     * Migrate N partitions
     *
     * shard X
     *  Y new RNs: rg7-rn1 rg7-rn2 rg7-rn3
     *  Z partition migrations
     *
     *  if verbose, a listing of all the partitions to migrated follows, sorted
     *  by source rep group, and then by partition id.
     */
    public String display(boolean verbose) {
        StringBuilder sb = new StringBuilder();
        /* Summary information. */
        int nShards = getNumNewShards();
        if (nShards > 0) {
            sb.append("Create ").append(nShards).
                append(" shard").append(plural(nShards)).append("\n");
        }

        int nNewRNs = getNumNewRNs();
        if (nNewRNs > 0) {
            sb.append("Create ").append(nNewRNs).append(" RN").
                append(plural(nNewRNs)).append("\n");
        }

        int moved = getNumRelocatedRNs();
        if (moved > 0) {
            sb.append("Relocate ").append(moved).append(" RN").
                append(plural(moved)).append("\n");
        }

        if (numCreatedPartitions != 0) {
            sb.append("Create ").append(numCreatedPartitions).
                append(" partition").append(plural(numCreatedPartitions)).
                append("\n");
        }

        if (numPartitionMigrations != 0) {
            sb.append("Migrate ").append(numPartitionMigrations).
                append(" partition").append(plural(numPartitionMigrations)).
                append("\n");
        }

        /* If some summary information has been generated, add a newline */
        if (sb.length() > 0) {
            sb.append("\n");
        }
        String indent = "  ";

        for (Map.Entry<RepGroupId,ShardChange> s : changedShards.entrySet()) {
            sb.append("shard ").append(s.getKey()).append("\n");
            ShardChange c = s.getValue();
            int newCreates = c.newlyCreatedRNs.size();
            if (newCreates > 0) {
                sb.append(indent);
                sb.append(newCreates).append(" new RN").
                    append(plural(newCreates)).append(": ");

                for (RepNodeId rnId : c.newlyCreatedRNs) {
                    sb.append(rnId).append(" ");
                }
                sb.append("\n");
            }

            if (!c.getRelocatedRNs().isEmpty()) {
                for (RelocatedRN rel : c.getRelocatedRNs()) {
                    sb.append(indent).append(rel).append("\n");
                }
            }

            if (c.newPartitionCount != 0) {
                sb.append(indent).append(c.newPartitionCount).
                    append(" new partition").
                    append(plural(c.newPartitionCount)).append("\n");
            }

            int nMigrates = c.migrations.size();
            if (nMigrates > 0) {
                sb.append(indent).append(nMigrates).
                    append(" partition migration").append(plural(nMigrates)).
                    append("\n");
                if (verbose) {
                    for (RelocatedPartition p :
                             sortBySrcAndPart(c.migrations)) {
                        sb.append(indent).append(p).append("\n");
                    }
                }
            }
        }

        if (sb.length() == 0) {
            return NO_CHANGE;
        }
        return "Topology transformation from " + sourceName + " to " +
            candidate.getName() + ":\n" + sb.toString();
    }

    /**
     * Return an "s" if the value is > 1, so the plurality is right in the
     * display.
     */
    private String plural(int val) {
        return (val==1) ? " " : "s ";
    }

    /**
     * Sort the partition migrations first by source shard, and then by
     * partition id, for display.
     */
    private List<RelocatedPartition> sortBySrcAndPart
        (List<RelocatedPartition> migrations) {

        List<RelocatedPartition> relPart =
            new ArrayList<RelocatedPartition>(migrations);
        Collections.sort
            (relPart,
             new Comparator<RelocatedPartition>() {
                @Override
                public int compare(RelocatedPartition p1,
                                   RelocatedPartition p2) {
                    int p1Src = p1.getSourceShard().getGroupId();
                    int p2Src = p2.getSourceShard().getGroupId();

                    int shardDiff = p1Src - p2Src;
                    if (shardDiff != 0) {
                        return shardDiff;
                    }
                    return p1.getPartitionId().getPartitionId() -
                        p2.getPartitionId().getPartitionId();
                }});
        return relPart;
    }

    /**
     * Return information about shards that have changed. This includes shards
     * that are newly created.
     */
    public Map<RepGroupId, ShardChange> getChangedShards() {
        return changedShards;
    }

    /**
     * Return the different in the number of shards between the two topologies.
     */
    public static int shardDelta(Topology topoA, Topology topoB) {
        return topoA.getRepGroupMap().size() - topoB.getRepGroupMap().size();
    }

    /**
     * The actual work of doing the diff.
     * @throws InvalidTopologyException
     */
    private void doDiff(boolean validate) {

        /*
         * Check that the source and target topologies obey all rules, and
         * that source can be legitimately transformed into the target.
         * Validation may be turned off if TopologyDiff is being used solely
         * to display potential transitions.
         */
        if (validate) {
            Rules.validateTransition(source, candidate, params);
        }

        /*
         * The topology was validated earlier to ensure that R2 assumptions are
         * correct. Currently, we don't support contraction, and we assume that
         * all RNs in the source topology are present in the target
         * topology.
         *
         * Find the set of brand new RNs that are only in the target topology,
         * and need to be created.
         */
        Topology target = candidate.getTopology();
        Set<RepNodeId> originalRNs = source.getRepNodeIds();
        newRNs.addAll(target.getRepNodeIds());
        newRNs.removeAll(originalRNs);

        /* Group the brand new RNs by shard. */
        identifyNewRNs();

        /*
         * Identify all the RNs that have been relocated, grouping them
         * by shard.
         */
        identifyRelocatedRNs(originalRNs);

        /*
         * Find all the shards that did not exist before..
         */
        findNewShards();

        /* Find partitions that need to be moved or created. */
        findPartitionChanges();

        mountPointAssignments = candidate.getAllMountPoints();
    }

    /**
     * Find shards that have new RNs.
     */
    private void identifyNewRNs() {

        /* Identify all brand new RNs, grouping them by shard. */
        for (RepNodeId rnId : newRNs) {
            ShardChange change =
                getShardChange(new RepGroupId(rnId.getGroupId()));
            change.addNewRN(rnId);
        }
    }

    /**
     * Add information about which shards have relocated RNs
     */
    private void identifyRelocatedRNs(Set<RepNodeId> originalRNs) {
        Topology target = candidate.getTopology();
        for (RepNodeId existingRNId : originalRNs) {
            StorageNodeId sourceSNId =
                source.get(existingRNId).getStorageNodeId();
            StorageNodeId targetSNId =
                target.get(existingRNId).getStorageNodeId();

            if (!sourceSNId.equals(targetSNId)) {
                String sourceMountPoint = null;
                RepNodeParams rnp = params.get(existingRNId);
                if (rnp != null) {
                    sourceMountPoint = rnp.getMountPointString();
                }
                String targetMountPoint =
                    candidate.getMountPoint(existingRNId);
                RelocatedRN rel = new RelocatedRN(existingRNId,
                                                  sourceSNId,
                                                  sourceMountPoint,
                                                  targetSNId,
                                                  targetMountPoint,
                                                  params);

                ShardChange change =
                    getShardChange(new RepGroupId(existingRNId.getGroupId()));
                change.addRelocatedRN(rel);
                numRelocatedRNs++;
            }
        }
    }

    /**
     *  Find all the shards which did not exist before.
     */
    private void findNewShards() {
        Topology target = candidate.getTopology();
        Set<RepGroupId> originalShards = source.getRepGroupIds();
        Set<RepGroupId> candidateShards = target.getRepGroupIds();

        Set<RepGroupId> brandNew = new HashSet<RepGroupId>(candidateShards);
        brandNew.removeAll(originalShards);
        newShards.addAll(brandNew);

        /*
         * Sort the new shard list so it's ordered by RepGroupId. This will
         * make the actual initial creation of partitions in the AddPartition
         * task match the candidate topology. New partition creation is
         * driven by a count of partitions on the ordinal shard rather than
         * an actual match of partition ids and shard ids, because the
         * shard id and partition id are generated internally by the topology,
         * and can't be specified by externally when constructing those items.
         */
        Collections.sort(newShards);

        removedShards.addAll(originalShards);
        removedShards.removeAll(candidateShards);

        /*
         * Just for R2, sanity check that there are no removed shards. This
         * should also have been checked via Rules.validateTransition, so this
         * is just a paranoid double-check.
         */
        if (removedShards.size() > 0) {
            throw new UnsupportedOperationException
                ("These shards " +  removedShards +
                 " are not present in the topology candidate, which means" +
                 " the candidate store is smaller, but store contraction" +
                 " is not supported in this release.");
        }
    }

    /**
     * Assumes that the newShards list has been populated before this is
     * called, as it assigns partitions to shards..
     */
    private void findPartitionChanges() {
        Topology target = candidate.getTopology();

        if (source.getPartitionMap().size() == 0) {
            /* Brand new store, assign partitions to new shards. */
            for (Partition p: target.getPartitionMap().getAll()) {
                ShardChange change = getShardChange(p.getRepGroupId());
                change.incNewPartionCount();
                numCreatedPartitions++;
            }
            return;
        }

        /*
         * Before looking at partition assignments, set aside RNs that already
         * have the right number of partitions. Assume that all partitions
         * are the same, and that we don't want to move partitions on or off
         * a RN if it already houses the right number. This guards against
         * inadvertent, unnecessary movement of partitions. In future releases,
         * we may give partitions different weight, and there may be a reason
         * to move a partitions even though the number of partitions on the
         * source or destination are unchanged.
         */

        /* Count the number of partitions per shard in the old topo */
        Map<RepGroupId, AtomicInteger> sourceCount =
            new HashMap<RepGroupId, AtomicInteger>();
        for (RepGroupId rgId: source.getRepGroupIds()) {
            sourceCount.put(rgId, new AtomicInteger(0));
        }
        for (Partition p: source.getPartitionMap().getAll()) {
            sourceCount.get(p.getRepGroupId()).incrementAndGet();
        }

        /* Count the number of partitions per shard in the new topo */
        Map<RepGroupId, AtomicInteger> targetCount =
            new HashMap<RepGroupId, AtomicInteger>();
        for (RepGroupId rgId: target.getRepGroupIds()) {
            targetCount.put(rgId, new AtomicInteger(0));
        }
        for (Partition p: target.getPartitionMap().getAll()) {
            targetCount.get(p.getRepGroupId()).incrementAndGet();
        }

        Set<RepGroupId> changed = new HashSet<RepGroupId>();
        for (RepGroupId rgId: target.getRepGroupIds()) {
            if (sourceCount.get(rgId) == null) {
                changed.add(rgId);
            } else if (sourceCount.get(rgId).get() !=
                       targetCount.get(rgId).get()) {
                changed.add(rgId);
            }
        }

        for (Partition p: target.getPartitionMap().getAll()) {
            PartitionId pId = p.getResourceId();
            RepGroupId oldShard = source.get(pId).getRepGroupId();
            RepGroupId newShard = target.get(pId).getRepGroupId();

            /*
             * If the shard assignments are different AND the old and new shard
             * changed the numbers of partitions they house, do a migration.
             */
            if ((!oldShard.equals(newShard)) &&
                (changed.contains(oldShard) && changed.contains(newShard))){

                ShardChange change = getShardChange(newShard);
                change.add(new RelocatedPartition(pId, oldShard));
                pMigrationSources.add(oldShard);
                pMigrationDestinations.add(newShard);
                numPartitionMigrations++;
            }
        }
    }

    /**
     * Get the appropriate ShardChange instance, creating one of it does not
     * already exist.
     */
    public ShardChange getShardChange(RepGroupId id) {
        ShardChange change = changedShards.get(id);
        if (change == null) {
            change = new ShardChange();
            changedShards.put(id, change);
        }
        return change;
    }

    /**
     * @return a list of newly created shards.
     */
    public List<RepGroupId> getNewShards() {
        return newShards;
    }

    /**
     * Records the changes that have to be made to a shard; which is to add or
     * move RNs, and add partitions.
     */
    public static class ShardChange {

        /* RNs that need to be created */
        private final Set<RepNodeId> newlyCreatedRNs;

        /* RNs that are to be moved from one SN to another. */
        private final Set<RelocatedRN> relocatedRNs;

        /*
         * The number of partitions that should be created on this shard.
         * Applies only to brand new shards. Partitions that are migrated
         * are recorded as a PartitionMigration.
         */
        private int newPartitionCount;

        /*
         * Partitions that are to be migrated to this shard.
         */
        private final List<RelocatedPartition> migrations;

        ShardChange() {

            /*
             * Keep the the relocated and newly created RNs sorted by node id,
             * so that the display of the diff is nicer. Also makes task
             * construction and order more deterministic and intuitive to the
             * user.
             */
            relocatedRNs = new TreeSet<RelocatedRN>
            (new Comparator<RelocatedRN>() {
                @Override
                public int compare(RelocatedRN o1, RelocatedRN o2) {
                    return o1.getRnId().getNodeNum() -
                        o2.getRnId().getNodeNum();
                }

            });
            newlyCreatedRNs = new TreeSet<RepNodeId>
            (new Comparator<RepNodeId>() {
                @Override
                public int compare(RepNodeId o1, RepNodeId o2) {
                    return o1.getNodeNum() - o2.getNodeNum();
                }

            });
            migrations = new ArrayList<RelocatedPartition>();
        }

        public List<RelocatedPartition> getMigrations() {
            return migrations;
        }

        public void add(RelocatedPartition relocatedPartition) {
            migrations.add(relocatedPartition);

        }

        void incNewPartionCount() {
            newPartitionCount++;
        }

        public int getNumNewPartitions() {
            return newPartitionCount;
        }

        public Set<RepNodeId> getNewRNs() {
            return newlyCreatedRNs;
        }

        void addNewRN(RepNodeId rnId) {
            newlyCreatedRNs.add(rnId);
        }

        public Set<RelocatedRN> getRelocatedRNs() {
            return relocatedRNs;
        }

        void addRelocatedRN(RelocatedRN rel) {
            relocatedRNs.add(rel);
        }

        /**
         * Make a string that contains the set of SNs that house this shard,
         * purely for debugging/descriptive purposes.
         */
        public String getSNSetDescription(Topology topo) {
            StringBuilder sb = new StringBuilder();
            for (RepNodeId rnId : newlyCreatedRNs) {
                StorageNodeId snId = topo.get(rnId).getStorageNodeId();
                sb.append(topo.get(snId).toString());
                sb.append(" ");
            }
            return sb.toString();
        }
    }

    /**
     * Records info for RNs that have moved to a different SN.
     */
    public static class RelocatedRN {
        private final RepNodeId rnId;
        private final StorageNodeId oldSNId;
        private final StorageNodeId newSNId;
        private final String oldMountPoint;
        private final String newMountPoint;
        private final String oldRootDir;
        private final String newRootDir;

        RelocatedRN(RepNodeId rnId,
                    StorageNodeId oldSNId,
                    String oldMountPoint,
                    StorageNodeId newSNId,
                    String newMountPoint,
                    Parameters params) {
            this.rnId = rnId;
            this.oldSNId = oldSNId;
            this.oldMountPoint = oldMountPoint;
            this.oldRootDir = params.get(oldSNId).getRootDirPath();
            this.newSNId = newSNId;
            this.newMountPoint = newMountPoint;
            this.newRootDir = params.get(newSNId).getRootDirPath();
        }

        public RepNodeId getRnId() {
            return rnId;
        }

        public StorageNodeId getNewSNId() {
            return newSNId;
        }

        public StorageNodeId getOldSNId() {
            return oldSNId;
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();
            sb.append("Relocate ").append(rnId).append(" from ");
            sb.append(oldSNId).append(" ");
            if (oldMountPoint == null) {
                sb.append(oldRootDir);
            } else {
                sb.append(oldMountPoint);
            }
            sb.append(" to ");
            sb.append(newSNId).append(" ");
            if (newMountPoint == null) {
                sb.append(newRootDir);
            } else {
                sb.append(newMountPoint);
            }
            return sb.toString();
        }
    }

    /**
     * Records info for partitions that have moved to a new shard.
     */
    public static class RelocatedPartition {

        private final PartitionId partitionId;
        private final RepGroupId sourceShard;

        public RelocatedPartition(PartitionId partitionId,
                                  RepGroupId sourceShard) {
            this.partitionId = partitionId;
            this.sourceShard = sourceShard;
        }

        public PartitionId getPartitionId() {
            return partitionId;
        }

        public RepGroupId getSourceShard() {
            return sourceShard;
        }

        @Override
        public String toString() {
            return "Migrate " + partitionId + " from " + sourceShard;
        }
    }

    public String getMountPoint(RepNodeId rnId) {
        return mountPointAssignments.get(rnId);
    }
}