/*
 * 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.cassandra.locator;

import java.io.InputStream;
import java.net.InetAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndpointState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ResourceWatcher;
import org.apache.cassandra.utils.WrappedRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.TypeDescription;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;

import com.google.common.base.Objects;
import com.google.common.net.InetAddresses;

/**
 * Network topology snitch that reads its configuration from a YAML file.
 * <p>
 * This snitch supports connections over preferred addresses, such as a data-center-local address, based on the
 * reconnection trick used in {@link Ec2MultiRegionSnitch}. The configuration file, {@code cassandra-topology.yaml}, is
 * checked periodically for updates.
 * </p>
 */
public class YamlFileNetworkTopologySnitch
        extends AbstractNetworkTopologySnitch
{

    private static final Logger logger = LoggerFactory.getLogger(YamlFileNetworkTopologySnitch.class);
    
    /**
     * How often to check the topology configuration file, in milliseconds; defaults to one minute.
     */
    private static final int CHECK_PERIOD_IN_MS = 60 * 1000;

    /** Default name for the topology configuration file. */
    private static final String DEFAULT_TOPOLOGY_CONFIG_FILENAME = "cassandra-topology.yaml";

    /** Node data map, keyed by broadcast address. */
    private volatile Map<InetAddress, NodeData> nodeDataMap;

    /** Node data for this node. */
    private volatile NodeData localNodeData;

    /** Node data to fall back to when there is no match. */
    private volatile NodeData defaultNodeData;

    /** Name of the topology configuration file. */
    private final String topologyConfigFilename;

    /** True if the gossiper has been initialized. */
    private volatile boolean gossiperInitialized = false;

    /**
     * Constructor.
     *
     * @throws ConfigurationException
     *             on failure
     */
    public YamlFileNetworkTopologySnitch() throws ConfigurationException
    {
        this(DEFAULT_TOPOLOGY_CONFIG_FILENAME);
    }

    /**
     * Constructor.
     *
     * @param topologyConfigFilename
     *            name of the topology configuration file
     * @throws ConfigurationException
     *             on failure
     */
    YamlFileNetworkTopologySnitch(final String topologyConfigFilename)
            throws ConfigurationException
    {
        this.topologyConfigFilename = topologyConfigFilename;
        loadTopologyConfiguration(false);

        try
        {
            /*
             * Check if the topology configuration file is a plain file.
             */
            FBUtilities.resourceToFile(topologyConfigFilename);

            final Runnable runnable = new WrappedRunnable()
            {
                /**
                 * Loads the topology.
                 */
                protected void runMayThrow() throws ConfigurationException
                {
                    loadTopologyConfiguration(true);
                }
            };
            ResourceWatcher.watch(topologyConfigFilename, runnable,
                    CHECK_PERIOD_IN_MS);
        }
        catch (final ConfigurationException e)
        {
            logger.debug(
                    "{} found, but does not look like a plain file. Will not watch it for changes",
                    topologyConfigFilename);
        }
    }

    /**
     * Returns the name of the rack for the endpoint, or {@code UNKNOWN} if not known.
     *
     * @return the name of the data center for the endpoint, or {@code UNKNOWN} if not known
     */
    @Override
    public String getRack(final InetAddress endpoint)
    {
        final NodeData nodeData = nodeDataMap.get(endpoint);
        return nodeData != null ? nodeData.rack : defaultNodeData.rack;
    }

    /**
     * Returns the name of the data center for the endpoint, or {@code UNKNOWN} if not known.
     *
     * @return the name of the data center for the endpoint, or {@code UNKNOWN} if not known
     */
    @Override
    public String getDatacenter(final InetAddress endpoint)
    {
        final NodeData nodeData = nodeDataMap.get(endpoint);
        return nodeData != null ? nodeData.datacenter
                : defaultNodeData.datacenter;
    }

    /**
     * Root object type for the YAML topology configuration.
     */
    public static class TopologyConfig
    {
        public List<Datacenter> topology;
        public String default_dc_name = "UNKNOWN";
        public String default_rack_name = "UNKNOWN";
    }

    /**
     * Data center object type for the YAML topology configuration.
     */
    public static class Datacenter
    {
        public String dc_name;
        public List<Rack> racks = Collections.emptyList();
    }

    /**
     * Rack object type for the YAML topology configuration.
     */
    public static class Rack
    {
        public String rack_name;
        public List<Node> nodes = Collections.emptyList();
    }

    /**
     * Node object type for the YAML topology configuration.
     */
    public static class Node
    {
        public String broadcast_address;
        public String dc_local_address;
    }

    /**
     * Loads the topology configuration file.
     *
     * @throws ConfigurationException
     *             on failure
     */
    private synchronized void loadTopologyConfiguration(boolean isUpdate)
            throws ConfigurationException
    {
        logger.debug("Loading topology configuration from {}",
                topologyConfigFilename);

        final TypeDescription topologyConfigTypeDescription = new TypeDescription(
                TopologyConfig.class);
        topologyConfigTypeDescription.putListPropertyType("topology",
                Datacenter.class);

        final TypeDescription topologyTypeDescription = new TypeDescription(
                Datacenter.class);
        topologyTypeDescription.putListPropertyType("racks", Rack.class);

        final TypeDescription rackTypeDescription = new TypeDescription(
                Rack.class);
        rackTypeDescription.putListPropertyType("nodes", Node.class);

        final Constructor configConstructor = new Constructor(
                TopologyConfig.class);
        configConstructor.addTypeDescription(topologyConfigTypeDescription);
        configConstructor.addTypeDescription(topologyTypeDescription);
        configConstructor.addTypeDescription(rackTypeDescription);

        final InputStream configFileInputStream = getClass().getClassLoader()
                .getResourceAsStream(topologyConfigFilename);
        if (configFileInputStream == null)
        {
            throw new ConfigurationException(
                    "Could not read topology config file "
                            + topologyConfigFilename);
        }
        Yaml yaml;
        TopologyConfig topologyConfig;
        try
        {
            yaml = new Yaml(configConstructor);
            topologyConfig = (TopologyConfig) yaml.load(configFileInputStream);
        }
        finally
        {
            FileUtils.closeQuietly(configFileInputStream);
        }
        final Map<InetAddress, NodeData> nodeDataMap = new HashMap<InetAddress, NodeData>();

        if (topologyConfig.topology == null)
        {
            throw new ConfigurationException(
                    "Topology configuration file is missing the topology section");
        }

        for (final Datacenter datacenter : topologyConfig.topology)
        {
            if (datacenter.dc_name == null)
            {
                throw new ConfigurationException(
                        "Topology configuration file is missing a data center name for some data center");
            }

            for (final Rack rack : datacenter.racks)
            {
                if (rack.rack_name == null)
                {
                    throw new ConfigurationException(
                            String.format(
                                    "Topology configuration file is missing a rack name for some rack under data center '%s'",
                                    datacenter.dc_name));
                }

                for (final Node node : rack.nodes)
                {
                    if (node.broadcast_address == null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "Topology configuration file is missing a broadcast address for some node under data center '%s' rack '%s'",
                                        datacenter.dc_name, rack.rack_name));
                    }

                    final InetAddress endpoint = InetAddresses
                            .forString(node.broadcast_address);
                    final InetAddress dcLocalAddress = node.dc_local_address == null ? null
                            : InetAddresses.forString(node.dc_local_address);

                    final NodeData nodeData = new NodeData();
                    nodeData.datacenter = datacenter.dc_name;
                    nodeData.rack = rack.rack_name;
                    nodeData.dcLocalAddress = dcLocalAddress;

                    if (nodeDataMap.put(endpoint, nodeData) != null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "IP address '%s' appears more than once in the topology configuration file",
                                        endpoint));
                    }

                    if (dcLocalAddress != null
                            && nodeDataMap.put(dcLocalAddress, nodeData) != null)
                    {
                        throw new ConfigurationException(
                                String.format(
                                        "IP address '%s' appears more than once in the topology configuration file",
                                        dcLocalAddress));
                    }
                }
            }
        }

        final NodeData localNodeData = nodeDataMap.get(FBUtilities
                .getBroadcastAddress());
        if (localNodeData == null)
        {
            throw new ConfigurationException(
                    "Topology configuration missing information for the local node");
        }

        final NodeData defaultNodeData = new NodeData();

        if (topologyConfig.default_dc_name == null)
        {
            throw new ConfigurationException(
                    "default_dc_name must be specified");
        }
        if (topologyConfig.default_rack_name == null)
        {
            throw new ConfigurationException(
                    "default_rack_name must be specified");
        }

        defaultNodeData.datacenter = topologyConfig.default_dc_name;
        defaultNodeData.rack = topologyConfig.default_rack_name;

        // YAML configuration looks good; now make the changes

        this.nodeDataMap = nodeDataMap;
        this.localNodeData = localNodeData;
        this.defaultNodeData = defaultNodeData;
        maybeSetApplicationState();

        if (logger.isDebugEnabled())
        {
            logger.debug(
                    "Built topology map from config file: localNodeData={}, nodeDataMap={}",
                    localNodeData, nodeDataMap);
        }

        if (gossiperInitialized)
        {
            StorageService.instance.gossipSnitchInfo();
        }

        if (isUpdate && StorageService.instance != null)
            StorageService.instance.updateTopology();
    }

    /**
     * be careful about just blindly updating ApplicationState.INTERNAL_IP everytime we read the yaml file,
     * as that can cause connections to get unnecessarily reset (via IESCS.onChange()).
     */
    private void maybeSetApplicationState()
    {
        if (localNodeData.dcLocalAddress == null)
            return;
        final EndpointState es = Gossiper.instance.getEndpointStateForEndpoint(FBUtilities.getBroadcastAddress());
        if (es == null)
            return;
        final VersionedValue vv = es.getApplicationState(ApplicationState.INTERNAL_IP);
        if ((vv != null && !vv.value.equals(localNodeData.dcLocalAddress.getHostAddress()))
            || vv == null)
        {
            Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP,
                StorageService.instance.valueFactory.internalIP(localNodeData.dcLocalAddress.getHostAddress()));
        }
    }

    /**
     * Topology data for a node.
     */
    private class NodeData
    {
        /** Data center name. */
        public String datacenter;
        /** Rack name. */
        public String rack;
        /** Data-center-local address. */
        public InetAddress dcLocalAddress;

        /**
         * Returns a simple key-value string representation of this node's data.
         *
         * @return a simple key-value string representation of this node's data
         */
        public String toString()
        {
            return Objects.toStringHelper(this).add("datacenter", datacenter)
                    .add("rack", rack).add("dcLocalAddress", dcLocalAddress)
                    .toString();
        }
    }

    /**
     * Called in preparation for the initiation of the gossip loop.
     */
    @Override
    public synchronized void gossiperStarting()
    {
        gossiperInitialized = true;
        StorageService.instance.gossipSnitchInfo();
        Gossiper.instance.register(new ReconnectableSnitchHelper(this, localNodeData.datacenter, true));
    }

}