/*
 * The contents of this file are subject to the terms
 * of the Common Development and Distribution License
 * (the License).  You may not use this file except in
 * compliance with the License.
 *
 * You can obtain a copy of the license at
 * https://glassfish.dev.java.net/public/CDDLv1.0.html or
 * glassfish/bootstrap/legal/CDDLv1.0.txt.
 * See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL
 * Header Notice in each file and include the License file
 * at glassfish/bootstrap/legal/CDDLv1.0.txt.
 * If applicable, add the following below the CDDL Header,
 * with the fields enclosed by brackets [] replaced by
 * you own identifying information:
 * "Portions Copyrighted [year] [name of copyright owner]"
 *
 * Copyright (c) Ericsson AB, 2004-2007. All rights reserved.
 * Copyright 1997-2007 Sun Microsystems, Inc. All rights reserved.
 */
package org.jvnet.glassfish.comms.clb.core;

import com.sun.enterprise.admin.server.core.AdminService;
import com.sun.enterprise.config.ConfigContext;
import com.sun.enterprise.config.ConfigException;
import com.sun.enterprise.config.serverbeans.Cluster;
import com.sun.enterprise.config.serverbeans.ClusterHelper;
import com.sun.enterprise.config.serverbeans.Config;
import com.sun.enterprise.config.serverbeans.ConfigAPIHelper;
import com.sun.enterprise.ee.cms.core.CallBack;
import com.sun.enterprise.ee.cms.core.FailureNotificationActionFactory;
import com.sun.enterprise.ee.cms.core.FailureNotificationSignal;
import com.sun.enterprise.ee.cms.core.GMSException;
import com.sun.enterprise.ee.cms.core.GMSFactory;
import com.sun.enterprise.ee.cms.core.GMSNotEnabledException;
import com.sun.enterprise.ee.cms.core.GroupHandle;
import com.sun.enterprise.ee.cms.core.GroupManagementService;
import com.sun.enterprise.ee.cms.core.JoinNotificationActionFactory;
import com.sun.enterprise.ee.cms.core.JoinNotificationSignal;
import com.sun.enterprise.ee.cms.core.PlannedShutdownActionFactory;
import com.sun.enterprise.ee.cms.core.PlannedShutdownSignal;
import com.sun.enterprise.ee.cms.core.ServiceProviderConfigurationKeys;
import com.sun.enterprise.ee.cms.core.Signal;
import com.sun.enterprise.ee.cms.impl.client.FailureNotificationActionFactoryImpl;
import com.sun.enterprise.ee.cms.impl.client.JoinNotificationActionFactoryImpl;
import com.sun.enterprise.ee.cms.impl.client.PlannedShutdownActionFactoryImpl;
import com.sun.enterprise.server.ApplicationServer;

import org.jvnet.glassfish.comms.util.LogUtil;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;

import java.net.Socket;
import java.net.UnknownHostException;

import java.util.List;
import java.util.Properties;
import java.util.logging.Level;


/**
 *
 * @author kshitiz
 */
public class ClusterHealthMonitor implements EventListener, CallBack {
    private static final LogUtil _logger = new LogUtil(ClusterHealthMonitor.class);
    private Controller controller;
    private String clusterName;
    private GroupManagementService gms;
    private FailureNotificationActionFactory failureNotificationActionFactory;
    private JoinNotificationActionFactory joinNotificationActionFactory;
    private PlannedShutdownActionFactory plannedShutdownActionFactory;

    /** Creates a new instance of ServerHealthMonitor */
    public ClusterHealthMonitor(Controller controller, ServerCluster cluster)
        throws CLBRuntimeException {
        this.clusterName = cluster.getName();
        this.controller = controller;
        gms = null;
        registerCallBackHandler(cluster);
    }

    public void processNotification(Signal signal) {
        if (signal instanceof JoinNotificationSignal) {
            onRecovery(signal.getGroupName(), signal.getMemberToken());
        } else if (signal instanceof FailureNotificationSignal) {
            onFailure(signal.getGroupName(), signal.getMemberToken());
        } else if (signal instanceof PlannedShutdownSignal) {
            onFailure(signal.getGroupName(), signal.getMemberToken());
        }
    }

    private void registerCallBackHandler(ServerCluster cluster)
        throws CLBRuntimeException {
        boolean startedGMSModule = false;
        while(gms == null){
            try {
                gms = GMSFactory.getGMSModule(clusterName);
            } catch (GMSNotEnabledException ex) {
                // It means gms module is not started for this cluster
                
                //For a self-loadbalancing cluster, this should already be started.
                //Throw an exception in this case
                if (cluster.isSelfLoadbalancing()) {
                    throw new CLBRuntimeException(
                            "The GMS service for self-loadbalancing cluster" +
                            " should be started at server startup.");
                }
                
                if(startedGMSModule)
                    throw new CLBRuntimeException("Already started GMS module for cluster : " + clusterName
                            + ". Still unable to get GMS module.");
                
                // Let me start a GMS module in this case
                _logger.logMsg(Level.INFO,
                        "Starting cluster gms module for cluster : " + clusterName);
                startGMSModule();
                startedGMSModule = true;
                _logger.logMsg(Level.INFO,
                        "Started cluster gms module for cluster : " + clusterName + ", GMS module : " + gms);
            } catch (GMSException ex) {
                throw new CLBRuntimeException(ex);
            }
        }

        failureNotificationActionFactory = new FailureNotificationActionFactoryImpl(this);
        gms.addActionFactory(failureNotificationActionFactory);

        joinNotificationActionFactory = new JoinNotificationActionFactoryImpl(this);
        gms.addActionFactory(joinNotificationActionFactory);

        plannedShutdownActionFactory = new PlannedShutdownActionFactoryImpl(this);
        gms.addActionFactory(plannedShutdownActionFactory);
    }

    private GroupManagementService startGMSModule() {
        String instanceName = com.sun.enterprise.server.ondemand.OnDemandServer.getServerContext()
                                                                               .getInstanceName();
        Runnable gmsModule = GMSFactory.startGMSModule(instanceName,
                clusterName, //Adding itself as spectator
            GroupManagementService.MemberType.SPECTATOR, getGMSConfigProps());
        
        ClusterGMSModuleThread gmsModuleThread = new ClusterGMSModuleThread(gmsModule);
        gmsModuleThread.start();
        
        return gms;
    }

    public List<String> getHealthyInstances() {
        GroupHandle groupHandle = gms.getGroupHandle();

        _logger.logMsg(Level.INFO,
            "All members : " + groupHandle.getAllCurrentMembers().toString());
        _logger.logMsg(Level.INFO,
            "All members : " + groupHandle.getCurrentCoreMembers().toString());
        //for all core cluster members (non spectators)
        return groupHandle.getCurrentCoreMembers();
    }

    public void onRecovery(String clusterName, String instanceName) {
        _logger.logMsg(Level.INFO,
            "Recieved recovery notification for instance : " + instanceName +
            " of cluster : " + clusterName);

        ServerInstance instance = getServerInstance(clusterName, instanceName);

        PingThread pingThread = new PingThread(instance);
        pingThread.start();
    }

    public void onFailure(String clusterName, String instanceName) {
        _logger.logMsg(Level.INFO,
            "Recieved failure notification for instance : " + instanceName +
            " of cluster : " + clusterName);

        ServerInstance instance = getServerInstance(clusterName, instanceName);
        instance.markAsUnhealthy();
    }

    public void onDisable(String clusterName, String instanceName) {
        ServerInstance instance = getServerInstance(clusterName, instanceName);
        instance.disableInstance();
    }

    public void onEnable(String clusterName, String instanceName) {
        ServerInstance instance = getServerInstance(clusterName, instanceName);
        instance.enableInstance();
    }

    private ServerInstance getServerInstance(String clusterName,
        String instanceName) {
        return controller.getInstance(clusterName, instanceName);
    }

    public String getClusterName() {
        return clusterName;
    }

    public void cleanup() {
        gms.removeActionFactory(failureNotificationActionFactory);
        gms.removeActionFactory(joinNotificationActionFactory);
        gms.removeActionFactory(plannedShutdownActionFactory);
    }

    private Properties getGMSConfigProps() {
        Properties props = new Properties();
        Config config;
        Cluster cluster;
        ConfigContext configContext;

        try {
            configContext = AdminService.getAdminService().getAdminContext()
                                        .getAdminConfigContext();
            cluster = ClusterHelper.getClusterByName(configContext, clusterName);

            String configRef = cluster.getConfigRef();
            config = ConfigAPIHelper.getConfigByName(configContext, configRef);

            com.sun.enterprise.config.serverbeans.GroupManagementService gmsConfig =
                config.getGroupManagementService();
            props.put(ServiceProviderConfigurationKeys.FAILURE_DETECTION_RETRIES.toString(),
                gmsConfig.getFdProtocolMaxTries());
            props.put(ServiceProviderConfigurationKeys.FAILURE_DETECTION_TIMEOUT.toString(),
                gmsConfig.getFdProtocolTimeoutInMillis());
            props.put(ServiceProviderConfigurationKeys.DISCOVERY_TIMEOUT.toString(),
                gmsConfig.getPingProtocolTimeoutInMillis());
            props.put(ServiceProviderConfigurationKeys.FAILURE_VERIFICATION_TIMEOUT.toString(),
                gmsConfig.getVsProtocolTimeoutInMillis());
            props.put(ServiceProviderConfigurationKeys.MULTICASTADDRESS.toString(),
                cluster.getHeartbeatAddress());
            props.put(ServiceProviderConfigurationKeys.MULTICASTPORT.toString(),
                cluster.getHeartbeatPort());
        } catch (ConfigException e) {
            _logger.logMsg(Level.SEVERE,
                "Exception when trying to properties for GMS module : " +
                e.getMessage());
            _logger.logMsg(Level.FINE, e, "Exception log : ");
        }

        return props;
    }

    class PingThread extends Thread {
        private ServerInstance instance;
        private final int maxRetryCount = 10;
        private final long sleepBetweenRetry = 5000L;

        PingThread(ServerInstance instance) {
            this.instance = instance;
        }

        public void run() {
            if (controller.getLBType() == CLBConstants.HTTP_CLB) {
                pingHttpEndPoint();
            } else {
                pingSipEndPoint();
            }
        }

        private void pingHttpEndPoint() {
            EndPoint endPoint;
            endPoint = instance.getEndPoint(CLBConstants.HTTP_PROTOCOL);

            if (endPoint == null) {
                endPoint = instance.getEndPoint(CLBConstants.HTTPS_PROTOCOL);
            }

            if (endPoint == null) {
                _logger.logMsg(Level.SEVERE,
                    "No endpoint to ping and determine status of instance");
            }

            sendHttpRequest(endPoint);
        }

        private void pingSipEndPoint() {
            //TBD
            //remove this dummy implementation
            //as of now it sleep for 30sec
            //and marks instance as healthy
            try {
                Thread.sleep(30000);
            } catch (InterruptedException ex) {
                _logger.logMsg(Level.SEVERE, ex, "Caught an exception");
            }

            _logger.logMsg(Level.INFO,
                "Marking instance " + instance.getName() + " of cluster " +
                clusterName + " as healthy");
            instance.markAsHealthy();
        }

        private void sendHttpRequest(EndPoint endPoint) {
            String request = "HEAD / HTTP/1.1\r\n" + "Host: " +
                endPoint.getHost() + ":" + endPoint.getPort() + "\r\n" +
                "Proxy-ping: true\r\n\r\n";
            int count = 0;

            while (count < maxRetryCount) {
                count++;

                try {
                    Socket socket = new Socket();
                    socket.connect(endPoint.getSocketAddress());

                    OutputStream outputStream = socket.getOutputStream();
                    outputStream.write(request.getBytes());
                    outputStream.flush();

                    BufferedReader reader = new BufferedReader(new InputStreamReader(
                                socket.getInputStream()));
                    String line = reader.readLine();

                    if (line != null) {
                        int index = line.indexOf(" ");

                        if (index > -1) {
                            int next = line.indexOf(" ", index + 1);

                            if (next > -1) {
                                int statusCode = Integer.valueOf(line.substring(index +
                                            1, next));

                                if ((statusCode >= 100) && (statusCode <= 499)) {
                                    _logger.logMsg(Level.INFO,
                                        "Ping request to instance " +
                                        instance.getName() + " of cluster " +
                                        clusterName + " is successful." +
                                        "Marking instance as healthy");
                                    instance.markAsHealthy();
                                }
                            }
                        }
                    }

                    try {
                        socket.close();
                    } catch (IOException ex) {
                        _logger.logMsg(Level.FINE, ex,
                            "Exception when closing socket. Ignoring");
                    }

                    if (instance.isHealthy()) {
                        break;
                    }

                    _logger.logMsg(Level.SEVERE,
                        "Instance : " + instance +
                        "is not healthy as response for HEAD request was incorrect. " +
                        "The response was : " + line);
                } catch (UnknownHostException ex) {
                    _logger.logMsg(Level.SEVERE, ex,
                        "Cannot open socket to instance : " +
                        instance.getName() + " of cluster " + clusterName +
                        " using host as " + endPoint.getHost() + " and port " +
                        endPoint.getPort());

                    break;
                } catch (IOException ex) {
                    _logger.logMsg(Level.SEVERE, ex,
                        "Exception when pinging instance : " +
                        instance.getName() + " of cluster " + clusterName);
                }

                try {
                    Thread.sleep(sleepBetweenRetry);
                } catch (InterruptedException ex) {
                    _logger.logMsg(Level.FINE, ex,
                        "Exception when sleeping between retry");
                }
            }

            if (!instance.isHealthy()) {
                _logger.logMsg(Level.SEVERE,
                    "Cannot mark instance : " + instance.getName() +
                    " of cluster " + clusterName + " as healthy");
            }
        }
    }
    
    class ClusterGMSModuleThread extends Thread{
        
        ClusterGMSModuleThread(Runnable gmsModule){
            super(gmsModule);
        }
        
    }
}
