Softec-Workflow / ProxyServer ADT

package it.softecspa.fileproxy.services;


import it.softecspa.database.dbconnect.ConnectionManager;
import it.softecspa.fileproxy.DatabaseBalancer;
import it.softecspa.fileproxy.db.ClusterInfo;
import it.softecspa.fileproxy.db.criterias.ClusterInfoCriteria;
import it.softecspa.fileproxy.services.ServerCacheFactory.PropertiesKey;
import it.softecspa.fileproxy.services.common.EnterpriseLog;
import it.softecspa.jwebber.frameworkImpl.security.RequestSecurityManager;
import it.softecspa.kahuna.lang.XString;
import it.softecspa.kahuna.net.SafeHttpURLConnection;
import it.softecspa.kahuna.util.Properties;
import it.softecspa.kahuna.util.calendar.EnterpriseCalendar;
import it.softecspa.portal.ApplicationClusterInfo;
import it.softecspa.portal.Parameters;
import it.softecspa.portal.Version;
import it.softecspa.portal.processRequest.HttpHeader;
import it.softecspa.portal.processRequest.xml.request.master.NodeInfo;
import it.softecspa.portal.processRequest.xml.request.master.NodeNotify;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.sql.SQLException;

import org.apache.log4j.Logger;

/**
 * Singleton per la gestione della sincronizzazione tra cluster
 * Le informazioni risiedono sul database STAGE
 * 
 * @author m.veroni
 *
 */
public class ClusterSynchronizer {
	
	// Singleton instance
	private static ClusterSynchronizer instance;
	
	
	private Logger log = Logger.getLogger(getClass());
	
	private ClusterInfo cluster;
	private ConnectionManager cmStage;
	
	private Thread heart;
	
	private ClusterSynchronizer() {
		super();
		synchronized (ClusterSynchronizer.class) {
			if (instance == null) {
				instance = this;
				if (log.isInfoEnabled()) log.info(ClusterSynchronizer.class.getSimpleName()+" instance is starting!");
				cmStage = DatabaseBalancer.getInstance().getStage();
			}
		}
	}

	public static ClusterSynchronizer getInstance() {
		synchronized (ClusterSynchronizer.class) {
			if (instance == null) new ClusterSynchronizer();			
		}
		return instance;
	}
	
	
	
	/**
	 * Implementazione del cuore che batte
	 * @author m.veroni
	 */
	public class Heart implements Runnable {

		@Override
		public void run() {		
			do {					
				// Se non è configurato alcun database non posso fare heartbeat
				if (!DatabaseBalancer.getInstance().isActive()) return;
				
				
				/*
				 * Parametro inserito nella configurazione su database
				 * In questo modo posso variare il "beat rate" a caldo
				 */
				try {
					int heartrate = ServerCacheFactory.getInstance().getDbProperties().getInt(PropertiesKey.HEARTBEAT_DELAY.toString(),60);
					if (heartrate<=0) {
						log.info("No hearthbeat!");
						return;
					}
					if (log.isDebugEnabled()) log.debug(">>>>>>> Heart beat delay: "+heartrate+" seconds");
				
				
					try {
						Thread.sleep(heartrate*1000);
					} catch (InterruptedException e) {
						log.warn("Heartbreak!");
						break;
					}				
					// Registro il battito
					heartbeat();
					
				} catch (Exception e) {
					log.error("Unhandled exception in heartbeat thread!",e);
				}
				
			} while (true);
			
		}

		
	}
	
	
	private enum MasterStageStatement {		
		
		  REGISTER("node-register")
		, BEAT("node-beat")
		, STOP("node-stop")
		, DEREGISTER("node-deregister");
		
		
		private String value;
		
		private MasterStageStatement(String value) {
			this.value = value;
		}

		public String getValue() {
			return value;
		}
		
	}
	
	
	
	
	public class RemoteCall implements Runnable {

		private URL url;
		private String username;
		private String password;
		
		private byte[] xmlInput;
		private long delay;
		
		
		public RemoteCall(URL url, String username, String password) {
			this(url,username,password,null,0);
		}
		
		public RemoteCall(URL url, String username, String password, byte[] xml, long delay) {
			this.url = url;
			this.username = username;
			this.password = password;
			//
			this.xmlInput = xml;
			this.delay = delay;
			
		}
		
		
		@Override
		public void run() {
			try {
				// -----------------------------------------------------------------------------
				
				if (delay>0) {
					log.warn("Delay comunication: "+delay+" seconds");
					try {
						Thread.sleep(delay*1000);
					} catch (InterruptedException e) {
						log.warn("Delay remote comunication!");
					}				
					
				}
				
				
				SafeHttpURLConnection connection;
				OutputStream output = null;	
				BufferedReader bufline = null;	
				try {
					connection = new SafeHttpURLConnection(url);
					connection.setRequestProperty(RequestSecurityManager.USERNAME, username);
					connection.setRequestProperty(RequestSecurityManager.PASSWORD, password);
					if (log.isDebugEnabled()) {
						if (username!=null || password!=null) {
							log.debug("Call url (with username/password) "+url.toString());
						} else {
							log.debug("Call url "+url.toString());
						}
					}
					
					connection.setRequestMethod("POST");
					connection.setRequestProperty("Content-Length",""+(xmlInput!=null?xmlInput.length:0));
					connection.setRequestProperty("Content-Type","text/xml");
					connection.setDoOutput(true); // posso scrivere!
					connection.setDoInput(true);  // posso leggere!
					connection.setUseCaches(false);
					
					// Comunico il mio XML a server remoto
					connection.write(xmlInput);
					
					// Analizza la risposta
					analyseResponse(connection);				
					
					// bufline = new BufferedReader(new InputStreamReader(connection.getInputStream()));
					if (log.isDebugEnabled()) log.debug("Remote call completed!");
					
					
				} catch (IOException e) {
					EnterpriseLog elog = new EnterpriseLog("IO error call url: "+url.toString(),e);
					elog.write();
				
				} catch (Exception e) {
					EnterpriseLog elog = new EnterpriseLog("Unhandled exception call url: "+url.toString(),e);
					elog.write();
				
				} catch (Error e) {
					EnterpriseLog elog = new EnterpriseLog("Unhandled error call url: "+url.toString(),e);
					elog.write();
					
				} finally {
					try {
						if (output!=null) output.close();
					} catch (IOException e) {
						/* nessuna operazione */
					}
					
					try {
						if (bufline!=null) bufline.close();
					} catch (IOException e) {
						/* nessuna operazione */
					}				
				}
				
			// -----------------------------------------------------------------------------	
			} catch (Exception e) {
				/* 
				 * Per risolvere il seguente problema in chiusura/undeploy della webapp:
				 * 
				 * 3-gen-2013 17.24.34 org.apache.catalina.loader.WebappClassLoader loadClass
				 * INFO: Illegal access: this web application instance has been stopped already.  Could not load it.softecspa.desktopmate.services.common.DatabaseLog.  The eventual following stack trace is caused by an error thrown for debugging purposes as well as to attempt to terminate the thread which caused the illegal access, and has no functional impact.
				 * java.lang.IllegalStateException
				 * 	at org.apache.catalina.loader.WebappClassLoader.loadClass(WebappClassLoader.java:1248)
				 *	at org.apache.catalina.loader.WebappClassLoader.loadClass(WebappClassLoader.java:1208)
				 *	at it.softecspa.desktopmate.services.ClusterSynchronizer$RemoteCall.run(ClusterSynchronizer.java:223)
				 *	at java.lang.Thread.run(Thread.java:662)
				 *
				 * Nota:
				 * http://www.javatuning.com/why-catch-throwable-is-evil-real-life-story/
				 */
				try {
					log.fatal("Very unhandled exception: "+e.toString(), e);
				} catch (Exception ei) {
					/* Nessuna operazione */
				} catch (Error ei) {
					/* Nessuna operazione */
				}
				
			} catch (Error e) {
				/*
				 * Vedi sopra
				 */
				try {
					log.fatal("Very unhandled error: "+e.toString(), e);
				} catch (Exception ei) {
					/* Nessuna operazione */
				} catch (Error ei) {
					/* Nessuna operazione */
				}
			}
		}


		/**
		 * Analizza la riposta del server remoto e se nagativa la inserisce nel database degli errori
		 * @param connection
		 */
		private void analyseResponse(SafeHttpURLConnection connection) {
			String url = "?";
			try {
				url = connection.getURL().toString();
			} catch (Exception e) {
				/* Nessuna operazione */
			}
			
			try {
				int httpCode = connection.getResponseCode();
				if (httpCode!=HttpURLConnection.HTTP_OK) { /* 200 */
					EnterpriseLog elog = new EnterpriseLog("Remote server http response error "+httpCode+" for url "+url,null);
					elog.write();
					return;
				}  				
				
			} catch (IOException e) {
				EnterpriseLog elog = new EnterpriseLog("Remote server IOException calling url "+url,e);
				elog.write();
				return;
			} catch (Exception e) {
				EnterpriseLog elog = new EnterpriseLog("Remote server Exception calling url "+url,e);
				elog.write();
				return;
			}
			
			
			if (log.isDebugEnabled()) log.debug("Analyze response header value");
			String result = connection.getHeaderField(HttpHeader.RESULT);
			String message = connection.getHeaderField(HttpHeader.MESSAGE);
			
			if (XString.isNotBlankNull(result)) {
				int res;
				try {
					res = Integer.parseInt(result);
				} catch (Exception e) {
					log.warn("Http header value '"+HttpHeader.RESULT+"' is not a number: '"+result+"'");
					return;
				}
				
				if (res<0) {
					// TODO prevedere una mail per esiti negativi delle chiamate remote
					
					EnterpriseLog elog = new EnterpriseLog("Remote server error response ["+res+"]: "+message,null);
					elog.write();
				}
			}
		}
		
	}
	
	
	public synchronized void init() throws Exception {		
		if (log.isInfoEnabled()) log.info("Initialize "+this.getClass().getSimpleName());
				
		// Recupero l'HostNameMasked da ApplicationClusterInfo
		ApplicationClusterInfo acInfo = ApplicationClusterInfo.getInstance();
		
		
		String public_hostname = acInfo.getPublicHostnameMasked();
		String backplane_hostname = acInfo.getBackplaneHostnameMasked();
		
		String context = Parameters.getInstance().get("application.context.name","/");
		if (!context.startsWith("/")) context = "/" + context;
		if (!context.endsWith("/")) context = context + "/";			
			
		
		
		try {			
			try {
				cluster = new ClusterInfo(cmStage, context, public_hostname);
			} catch (Exception e) {
				log.error("Impossible to recover host info from database",e);	
				if (cluster==null) cluster = new ClusterInfo();
			}
			/*
			 * Workaround per cambio nome cluster dev (popolo01)
			 */
			if (acInfo.getHostName().startsWith("ec2")) {
				log.warn("Found Amazon EC2 server name, mark do not use!");
				cluster.setDoNotUse(true);
			}
			
			
			cluster.setHostname(public_hostname);
			cluster.setContext(context);
			cluster.setAddress(acInfo.getHostAddress());
			cluster.setBackplaneHostname(backplane_hostname);
			cluster.setLifeStart(EnterpriseCalendar.now());
			cluster.setLastBeat(cluster.getLifeStart());
			cluster.setVersion(Version.getInstance().toString());
			cluster.lifeStart(cmStage);
			
		} catch (Exception e) {
			log.error("Impossible to save host info on database",e);
		}
		
		
		if (cluster.getDoNotUse()) {
			if (log.isInfoEnabled()) log.info("Heart is broken for this host named "+getPublicHostName());		
			// Notifica della deregistrazione del nodo al sistema master
			notify2Master(MasterStageStatement.DEREGISTER);
			
		} else {			
			heart = new Thread(new Heart(),"Heart "+getPublicHostName());
			heart.start();
			if (log.isInfoEnabled()) log.info("Heart start for "+getPublicHostName());
		
			// Notifica dello START al sistema master
			notify2Master(MasterStageStatement.REGISTER);
		}
	}

	
	
	
	/**
	 * Notifica al server master (se configurato) lo stato del server
	 * @param statement
	 */
	private void notify2Master(MasterStageStatement statement) {
		
		Properties prop = ServerCacheFactory.getInstance().getDbProperties();
		String server = prop.get(PropertiesKey.MASTER_SERVER_URL.toString());
		if (XString.isNotBlankNull(server)) {
			
			String username = prop.get(PropertiesKey.MASTER_SERVER_USERNAME.toString());
			String password = prop.get(PropertiesKey.MASTER_SERVER_PASSWORD.toString());
			
			if (log.isInfoEnabled()) log.info("Notify node info to master server (statement = '"+statement.getValue()+"') with url "+server);
			URL url = null;
			try {
				url = new URL(server+"?statement="+statement.getValue());
			} catch (MalformedURLException e) {				
				EnterpriseLog elog = new EnterpriseLog("Notify node info to master server "+server,e);
				elog.write();
				return;
			}
			Parameters parameters = Parameters.getInstance();
			
			
			
			
			// Composizione XML da trasmettere al server
			NodeInfo cluster_info = new NodeInfo();
			// Valido per uttti gli stati
			cluster_info.setHostname(cluster.getHostname());
			cluster_info.setContext(cluster.getContext());
			// solo START
			if (statement.equals(MasterStageStatement.REGISTER)) {
				cluster_info.setAddress(cluster.getAddress());
				cluster_info.setUrl(parameters.getChannelInfo().getDomainName() +
									parameters.getChannelInfo().getPortHTTP() + 
									parameters.getChannelInfo().getContextName());
				//
				cluster_info.setDm_version(cluster.getVersion());
				cluster_info.setJava_version(System.getProperty("java.version"));
				cluster_info.setOs_version(System.getProperty("os.name") + 
									  " ("+System.getProperty("os.arch")+") " +
										   System.getProperty("os.version"));
				//
				cluster_info.setPassword(prop.get(PropertiesKey.CLUSTER_USERNAME.toString()));
				cluster_info.setUsername(prop.get(PropertiesKey.CLUSTER_PASSWORD.toString()));				
				cluster_info.setStart(cluster.getLifeStart());
				//
				int heartrate = ServerCacheFactory.getInstance().getDbProperties().getInt(PropertiesKey.HEARTBEAT_DELAY.toString(),60);
				heartrate = heartrate*parameters.getInt(PropertiesKey.MASTER_HEARTBEAT_COUNT.toString(),1);				
				cluster_info.setBeat_delay(heartrate);
			} 
			// valido per START e BEAT
			if (statement.equals(MasterStageStatement.BEAT) || 
				statement.equals(MasterStageStatement.REGISTER)) {
				cluster_info.setBeat(cluster.getLastBeat());
			// solo per STOP
			} else if (statement.equals(MasterStageStatement.STOP)) {
				cluster_info.setStop(cluster.getLifeStop());
			}
			
			NodeNotify node = new NodeNotify();
			node.setAction(statement.getValue());
			node.setInfo(cluster_info);
			
			byte[] xml = null;
			try {
				xml = node.generateXML("UTF-8", log.isDebugEnabled());
				if (log.isTraceEnabled()) {
					log.debug("Slave to master trasmission:\n"+new String(xml));
				}
			} catch (UnsupportedEncodingException e) {				
				EnterpriseLog elog = new EnterpriseLog("Notify node info to master server "+server,e);
				elog.write();
				return;
			}
			
			// Aggiunto delay di 30 secondi di ritardo per evitare un errore in case di semplice riavvio
			if (log.isDebugEnabled()) log.debug("Start new thread for remote call to "+url);
			Thread master = new Thread(new RemoteCall(url, username, password, xml, 30),"Master notify to "+server);
			master.start();			
		}
		
		
	}

	/**
	 * Registra sul database che il nodo è stato fermato
	 */
	public void endLife() {
		if (log.isInfoEnabled()) log.info("Registered node life stop");
		EnterpriseCalendar adesso = EnterpriseCalendar.now();
		
		if (heart!=null) {
			if (heart.isAlive()) heart.interrupt();
		}
		
		if (cluster!=null) {
			cluster.setLifeStop(adesso);
			try {
				cluster.lifeStop(cmStage);
			} catch (SQLException e) {
				log.warn("Error in register life stop",e);
			}
			
			
			if (!cluster.getDoNotUse()) {
				// Notifica dello STOP al sistema master
				notify2Master(MasterStageStatement.STOP);
			}			
		}		
		
		
	}

	public void heartbeat() {
		if (log.isInfoEnabled()) log.info("Cluster hartbeat, now!");
		
		if (cluster!=null) {
			cluster.setLastBeat(EnterpriseCalendar.now());
			try {
				cluster.heartBeat(cmStage);
				
				
				// Comunico la cosa al server centrale am con periodicità diversa
				Properties prop = ServerCacheFactory.getInstance().getDbProperties();
				int count = prop.getInt(PropertiesKey.MASTER_HEARTBEAT_COUNT.toString(),1);
				if (cluster.addBeatCount()>=count) {
					cluster.resetBeatCount();					
					notify2Master(MasterStageStatement.BEAT);
				}
				
			} catch (SQLException e) {
				log.error("SQLException in register hartbeat",e);
			
			} catch (Exception e) {
				log.error("Unhandled exception in register hartbeat",e);
			}
		}		
	}
	
	
	public String getHostAddress() {
		return cluster.getAddress();
	}
	
	public String getPublicHostName() {
		return cluster.getHostname();
	}
	
	public String getBackplaneHostName() {
		return cluster.getBackplaneHostname();
	}

	
	
	
	/**
	 * Estrae la lista aggiornata di tutti i cluster attivi e 
	 * chiama il servizio remoto con i parametri richiesti
	 * E' utilizzata l'autenticazione
	 * @param requestURI
	 * @param statement
	 */
	public void statementReplicator(String servlet, String statement) {
		try {
			ClusterInfoCriteria criteria = new ClusterInfoCriteria(cmStage);
			criteria.setActiveOnly(Boolean.TRUE);
			ClusterInfo[] clusters = criteria.select();
			
			
			Properties prop = ServerCacheFactory.getInstance().getDbProperties();
			String username = prop.get(PropertiesKey.CLUSTER_USERNAME.toString());
			String password = prop.get(PropertiesKey.CLUSTER_PASSWORD.toString());
			
			
			for (ClusterInfo remote : clusters) {
				
				if (remote.getHostname().equals(this.cluster.getHostname())) continue;
				
				URL url = null;
				String hostname = (remote.getBackplaneHostname()!=null?remote.getBackplaneHostname():remote.getHostname());
				try {
					url = new URL("http://" + hostname + remote.getContext() + servlet +"?"+ statement);
				} catch (MalformedURLException e) {
					EnterpriseLog elog = new EnterpriseLog("Remote call on "+hostname,e);
					elog.write();
					continue;
				}
				
				if (log.isDebugEnabled()) log.debug("Start new thread for remote call to "+url);
				Thread thread = new Thread(new RemoteCall(url, username, password));
				thread.start();		
			}
			
		} catch (SQLException e) {
			log.error("SQLException reloading remote semaphore",e);
		} catch (Exception e) {
			log.error("Unhandled exception reloading remote semaphore",e);	
		}
	}

	
	
	
	
	/* FIXME da sistemare
	
	/**
	 * Ricarica la cache di sistema sui server remoti
	 * @param balancer
	 * @param tablename
	 * /
	private void loadRemoteCache(Balancer balancer, String tablename) {		
		if (log.isInfoEnabled()) {
			try {
				if (tablename!=null) {
					log.info("Launch remote cache reload "+(balancer!=null?"on "+balancer.toString()+" database":"")+" for table '"+tablename+"'");
				} else if (balancer!=null) {
					log.info("Launch remote cache reload on "+balancer.toString()+" database");
				} else {
					log.info("Launch remote cache reload");
				}
			} catch (Exception e) {
				log.warn("logger error",e);
			}
		}
		
		String statement = ServicesSynchroPR.RELOAD+"="+ServicesSynchroPR.RELOAD_LOCAL_CACHE + 
						   // Database	
		   				   (balancer!=null?"&"+ServicesSynchroPR.DATABASE+"="+(balancer.equals(Balancer.MASTER)?ServicesSynchroPR.DATABASE_MASTER:ServicesSynchroPR.DATABASE_STAGE):"") +
		   				   // Tablename
		   				   (tablename!=null?"&"+ServicesSynchroPR.TABLE_NAME+"="+tablename:"");
		statementReplicator(ServicesSynchroPR.SERVLET_NAME, statement);		
	}
	
	public void reloadRemoteSemaphore() {	
		String statement = ServicesSynchroPR.RELOAD+"="+ServicesSynchroPR.RELOAD_LOCAL_SEMAPHORE;
		ClusterSynchronizer.getInstance().statementReplicator(ServicesSynchroPR.SERVLET_NAME, statement);
	}
	
	
	
	public void reloadRemoteCacheOnMaster(String tablename) {
		loadRemoteCache(Balancer.MASTER, tablename);
	}
	
	public void reloadRemoteCacheOnMaster() {
		reloadRemoteCacheOnMaster(null);
	}
	
	public void reloadRemoteCacheOnStage(String tablename) {
		loadRemoteCache(Balancer.STAGE, tablename);
	}
	
	public void reloadRemoteCacheOnStage() {
		reloadRemoteCacheOnStage(null);
	}
	
	public void reloadRemoteCache() {
		loadRemoteCache(null, null);
	}
	
	public void reloadRemoteCache(String tablename) {
		loadRemoteCache(null, tablename);
	}
	
	*/
	
	
	
	
	
	
}