axis2/src/main/java/org/apache/ode/axis2/deploy/DeploymentPoller.java - ode - Git at Google

 /*
  * 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.
  */

 /*
  * 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.ode.axis2.deploy;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.ode.axis2.ODEServer;
 import org.apache.ode.bpel.clapi.ClusterLock;
 import org.apache.ode.bpel.engine.cron.CronScheduler;
 import org.apache.ode.bpel.engine.cron.SystemSchedulesConfig;
 import org.apache.ode.utils.WatchDog;

 import javax.xml.namespace.QName;
 import java.io.File;
 import java.io.FileFilter;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;

 /**
  * Polls a directory for the deployment of a new deployment unit.
  */
 public class DeploymentPoller {

     private static Logger __log = LoggerFactory.getLogger(DeploymentPoller.class);

     /** The polling interval. */
     private static final long POLL_TIME = 3000;

     private File _deployDir;

     private PollingThread _poller;

     protected ODEServer _odeServer;

     private boolean _onHold = false;

     private SystemSchedulesConfig _systemSchedulesConf;

     private boolean clusterEnabled;

     @SuppressWarnings("unchecked")
     private Map<String, WatchDog> dDWatchDogsByPath = new HashMap<String, WatchDog>();
     @SuppressWarnings("unchecked")
     private WatchDog _systemCronConfigWatchDog;

     /** Filter accepting directories containing a ode dd file. */
     private static final FileFilter _fileFilter = new FileFilter() {
         public boolean accept(File path) {
             return new File(path, "deploy.xml").exists();
         }
     };

     /** Filter accepting *.deployed files. */
     private static final FileFilter _deployedFilter = new FileFilter() {
         public boolean accept(File path) {
             return path.isFile() && path.getName().endsWith(".deployed");
         }
     };

     public DeploymentPoller(File deployDir, final ODEServer odeServer) {
         _odeServer = odeServer;
         _deployDir = deployDir;
         clusterEnabled = _odeServer.isClusteringEnabled();
         if (!_deployDir.exists()) {
             boolean isDeployDirCreated = _deployDir.mkdir();
             if (!isDeployDirCreated) {
                 __log.error("Error while creating deploy directory "
                         + deployDir.getName());
             }
         }
         _systemSchedulesConf = createSystemSchedulesConfig(odeServer.getConfigRoot());
         _systemCronConfigWatchDog = createSystemCronConfigWatchDog(odeServer.getBpelServer().getContexts().cronScheduler);
     }

     public void start() {
         _poller = new PollingThread();
         _poller.start();
         __log.info("Poller started.");
     }

     public void stop() {
         _poller.kill();
         _poller = null;
         __log.info("Poller stopped.");
     }

     protected boolean isDeploymentFromODEFileSystemAllowed() {
         return true;
     }

     /**
      * Scan the directory for new (or removed) files (called mainly from {@link PollingThread}) and calls whoever is in charge of
      * the actual deployment (or undeployment).
      */
     @SuppressWarnings("unchecked")
     private void check() {
         File[] files = _deployDir.listFiles(_fileFilter);
         boolean duLocked;

         // Checking for new deployment directories
         if (isDeploymentFromODEFileSystemAllowed() && files != null) {
             for (File file : files) {
                 String duName = file.getName();
                 if (__log.isDebugEnabled()) {
                 __log.debug("Trying to acquire the lock for " + duName);
                 }
                 duLocked = pollerTryLock(duName);

                 if (duLocked) {
                     try {
                         File deployXml = new File(file, "deploy.xml");
                         File deployedMarker = new File(_deployDir, file.getName() + ".deployed");

                         if (!deployXml.exists()) {
                             // Skip if deploy.xml is abset
                             if (__log.isDebugEnabled()) {
                                 __log.debug("Not deploying " + file + " (missing deploy.xml)");
                             }
                         }

                         WatchDog ddWatchDog = ensureDeployXmlWatchDog(file, deployXml);

                         if (deployedMarker.exists()) {
                             checkDeployXmlWatchDog(ddWatchDog);
                             continue;
                         }

                         try {
                             boolean isCreated = deployedMarker.createNewFile();
                             if (!isCreated) {
                                 __log.error("Error while creating  file "
                                         + file.getName()
                                         + ".deployed ,deployment could be inconsistent");
                             }
                         } catch (IOException e1) {
                             __log.error("Error creating deployed marker file, " + file + " will not be deployed");
                             continue;
                         }

                         try {
                             _odeServer.getProcessStore().undeploy(file);
                         } catch (Exception ex) {
                             __log.error("Error undeploying " + file.getName());
                         }

                         try {
                             Collection<QName> deployed = _odeServer.getProcessStore().deploy(file);
                             __log.info("Deployment of artifact " + file.getName() + " successful: " + deployed);
                         } catch (Exception e) {
                             __log.error("Deployment of " + file.getName() + " failed, aborting for now.", e);
                         }
                     } finally {
                         if (__log.isDebugEnabled()) {
                         __log.debug("Trying to release the lock for " + file.getName());
                         }
                         unlock(file.getName());
                     }
                 }
             }
         }

         // Removing deployments that disappeared
         File[] deployed = _deployDir.listFiles(_deployedFilter);
         for (File file : deployed) {
             String pkg = file.getName().substring(0, file.getName().length() - ".deployed".length());
             File deployDir = new File(_deployDir, pkg);
             if (!deployDir.exists()) {
                 String duName = deployDir.getName();

                 if (__log.isDebugEnabled()) {
                     __log.debug("Trying to acquire the lock for " + duName);
                 }

                 duLocked = pollerTryLock(duName);

                 if (duLocked) {
                     try {
                         Collection<QName> undeployed = _odeServer.getProcessStore().undeploy(deployDir);
                         boolean isDeleted = file.delete();
                         if (!isDeleted) {
                             __log.error("Error while deleting file "
                                     + file.getName()
                                     + ".deployed , please check if file is locked or if it really exist");
                         }
                         disposeDeployXmlWatchDog(deployDir);
                         if (undeployed.size() > 0)
                             __log.info("Successfully undeployed " + pkg);
                     } finally {
                         if (__log.isDebugEnabled()) {
                             __log.debug("Trying to release the lock for " + duName);
                         }
                         unlock(duName);
                     }
                 }
             }
         }

         checkSystemCronConfigWatchDog(_systemCronConfigWatchDog);
     }

     @SuppressWarnings("unchecked")
     protected WatchDog ensureDeployXmlWatchDog(File deployFolder, File deployXml) {
         WatchDog ddWatchDog = dDWatchDogsByPath.get(deployXml.getAbsolutePath());
         if( ddWatchDog == null ) {
             ddWatchDog = WatchDog.watchFile(deployXml, new DDWatchDogObserver(deployFolder.getName()));
             dDWatchDogsByPath.put(deployXml.getAbsolutePath(), ddWatchDog);
         }

         return ddWatchDog;
     }

     @SuppressWarnings("unchecked")
     protected void checkDeployXmlWatchDog(WatchDog ddWatchDog) {
         ddWatchDog.check();
     }

     protected void disposeDeployXmlWatchDog(File deployDir) {
         dDWatchDogsByPath.remove(new File(deployDir, "deploy.xml").getAbsolutePath());
     }

     protected SystemSchedulesConfig createSystemSchedulesConfig(File configRoot) {
         return new SystemSchedulesConfig(configRoot);
     }

     @SuppressWarnings("unchecked")
     protected WatchDog createSystemCronConfigWatchDog(final CronScheduler cronScheduler) {
         return WatchDog.watchFile(_systemSchedulesConf.getSchedulesFile(),
             new WatchDog.DefaultObserver() {
                 public void init() {
                     cronScheduler.refreshSystemCronJobs(_systemSchedulesConf);
                 }
             });
     }

     @SuppressWarnings("unchecked")
     protected void checkSystemCronConfigWatchDog(WatchDog ddWatchDog) {
         ddWatchDog.check();
     }

     /**
      * Thread that does the actual polling for new files.
      */
     private class PollingThread extends Thread {
         private boolean _active = true;

         public PollingThread() {
             setName("DeploymentPoller");
         }

         /** Stop this poller, and block until it terminates. */
         void kill() {
             synchronized (this) {
                 _active = false;
                 this.notifyAll();
             }
             try {
                 join();
             } catch (InterruptedException ie) {
                 __log.error("Thread unexpectedly interrupted.", ie);
             }
         }

         public void run() {
             try {
                 while (_active) {
                     if (!_onHold)
                         check();
                     synchronized (this) {
                         try {
                             this.wait(POLL_TIME);
                         } catch (InterruptedException e) {
                         }
                     }
                 }
             } catch (Throwable t) {
                 __log.error("Encountered an unexpected error.  Exiting poller...", t);
             }
         }
     }

     public void hold() {
         _onHold = true;
     }

     public void release() {
         _onHold = false;
     }

     public void markAsDeployed(File file) {
         File deployedMarker = new File(_deployDir, file.getName() + ".deployed");
         try {
             boolean isCreated = deployedMarker.createNewFile();
             if (!isCreated) {
                 __log.error("Error while creating  file " + file.getName()
                         + ".deployed ,deployment could be inconsistent");
             }
         } catch (IOException e) {
             __log.error("Couldn't create marker file for " + file.getName());
         }
     }

     public void markAsUndeployed(File file) {
         File deployedMarker = new File(_deployDir, file.getName() + ".deployed");
         boolean isDeleted = deployedMarker.delete();
         if (!isDeleted) {
             __log
                     .error("Error while deleting file "
                             + file.getName()
                             + ".deployed , please check if file is locked or if it really exist");
         }
     }

     @SuppressWarnings("unchecked")
     protected class DDWatchDogObserver extends WatchDog.DefaultObserver {
         private String deploymentPakage;

         public DDWatchDogObserver(String deploymentPakage) {
             this.deploymentPakage = deploymentPakage;
         }

         public void init() {
             _odeServer.getProcessStore().refreshSchedules(deploymentPakage);
         }
     }

     /**
      * Use to acquire the lock by poller
      */
     private boolean pollerTryLock(String key) {
         if(clusterEnabled) {
             ClusterLock clusterLock = _odeServer.getBpelServer().getContexts().clusterManager.getDeploymentLock();
             clusterLock.putIfAbsent(key,key);
             return clusterLock.tryLock(key);
         }
         else return true;
     }

     private boolean unlock(String key) {
         if (clusterEnabled) {
             _odeServer.getBpelServer().getContexts().clusterManager.getDeploymentLock().unlock(key);
         }
         return true;
     }
 }
	/*
	* 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.
	*/

	/*
	* 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.ode.axis2.deploy;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.apache.ode.axis2.ODEServer;
	import org.apache.ode.bpel.clapi.ClusterLock;
	import org.apache.ode.bpel.engine.cron.CronScheduler;
	import org.apache.ode.bpel.engine.cron.SystemSchedulesConfig;
	import org.apache.ode.utils.WatchDog;

	import javax.xml.namespace.QName;
	import java.io.File;
	import java.io.FileFilter;
	import java.io.IOException;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.Map;

	/**
	* Polls a directory for the deployment of a new deployment unit.
	*/
	public class DeploymentPoller {

	private static Logger __log = LoggerFactory.getLogger(DeploymentPoller.class);

	/** The polling interval. */
	private static final long POLL_TIME = 3000;

	private File _deployDir;

	private PollingThread _poller;

	protected ODEServer _odeServer;

	private boolean _onHold = false;

	private SystemSchedulesConfig _systemSchedulesConf;

	private boolean clusterEnabled;

	@SuppressWarnings("unchecked")
	private Map<String, WatchDog> dDWatchDogsByPath = new HashMap<String, WatchDog>();
	@SuppressWarnings("unchecked")
	private WatchDog _systemCronConfigWatchDog;

	/** Filter accepting directories containing a ode dd file. */
	private static final FileFilter _fileFilter = new FileFilter() {
	public boolean accept(File path) {
	return new File(path, "deploy.xml").exists();
	}
	};

	/** Filter accepting .deployed files. /
	private static final FileFilter _deployedFilter = new FileFilter() {
	public boolean accept(File path) {
	return path.isFile() && path.getName().endsWith(".deployed");
	}
	};

	public DeploymentPoller(File deployDir, final ODEServer odeServer) {
	_odeServer = odeServer;
	_deployDir = deployDir;
	clusterEnabled = _odeServer.isClusteringEnabled();
	if (!_deployDir.exists()) {
	boolean isDeployDirCreated = _deployDir.mkdir();
	if (!isDeployDirCreated) {
	__log.error("Error while creating deploy directory "
	+ deployDir.getName());
	}
	}
	_systemSchedulesConf = createSystemSchedulesConfig(odeServer.getConfigRoot());
	_systemCronConfigWatchDog = createSystemCronConfigWatchDog(odeServer.getBpelServer().getContexts().cronScheduler);
	}

	public void start() {
	_poller = new PollingThread();
	_poller.start();
	__log.info("Poller started.");
	}

	public void stop() {
	_poller.kill();
	_poller = null;
	__log.info("Poller stopped.");
	}

	protected boolean isDeploymentFromODEFileSystemAllowed() {
	return true;
	}

	/**
	* Scan the directory for new (or removed) files (called mainly from {@link PollingThread}) and calls whoever is in charge of
	* the actual deployment (or undeployment).
	*/
	@SuppressWarnings("unchecked")
	private void check() {
	File[] files = _deployDir.listFiles(_fileFilter);
	boolean duLocked;

	// Checking for new deployment directories
	if (isDeploymentFromODEFileSystemAllowed() && files != null) {
	for (File file : files) {
	String duName = file.getName();
	if (__log.isDebugEnabled()) {
	__log.debug("Trying to acquire the lock for " + duName);
	}
	duLocked = pollerTryLock(duName);

	if (duLocked) {
	try {
	File deployXml = new File(file, "deploy.xml");
	File deployedMarker = new File(_deployDir, file.getName() + ".deployed");

	if (!deployXml.exists()) {
	// Skip if deploy.xml is abset
	if (__log.isDebugEnabled()) {
	__log.debug("Not deploying " + file + " (missing deploy.xml)");
	}
	}

	WatchDog ddWatchDog = ensureDeployXmlWatchDog(file, deployXml);

	if (deployedMarker.exists()) {
	checkDeployXmlWatchDog(ddWatchDog);
	continue;
	}

	try {
	boolean isCreated = deployedMarker.createNewFile();
	if (!isCreated) {
	__log.error("Error while creating file "
	+ file.getName()
	+ ".deployed ,deployment could be inconsistent");
	}
	} catch (IOException e1) {
	__log.error("Error creating deployed marker file, " + file + " will not be deployed");
	continue;
	}

	try {
	_odeServer.getProcessStore().undeploy(file);
	} catch (Exception ex) {
	__log.error("Error undeploying " + file.getName());
	}

	try {
	Collection<QName> deployed = _odeServer.getProcessStore().deploy(file);
	__log.info("Deployment of artifact " + file.getName() + " successful: " + deployed);
	} catch (Exception e) {
	__log.error("Deployment of " + file.getName() + " failed, aborting for now.", e);
	}
	} finally {
	if (__log.isDebugEnabled()) {
	__log.debug("Trying to release the lock for " + file.getName());
	}
	unlock(file.getName());
	}
	}
	}
	}

	// Removing deployments that disappeared
	File[] deployed = _deployDir.listFiles(_deployedFilter);
	for (File file : deployed) {
	String pkg = file.getName().substring(0, file.getName().length() - ".deployed".length());
	File deployDir = new File(_deployDir, pkg);
	if (!deployDir.exists()) {
	String duName = deployDir.getName();

	if (__log.isDebugEnabled()) {
	__log.debug("Trying to acquire the lock for " + duName);
	}

	duLocked = pollerTryLock(duName);

	if (duLocked) {
	try {
	Collection<QName> undeployed = _odeServer.getProcessStore().undeploy(deployDir);
	boolean isDeleted = file.delete();
	if (!isDeleted) {
	__log.error("Error while deleting file "
	+ file.getName()
	+ ".deployed , please check if file is locked or if it really exist");
	}
	disposeDeployXmlWatchDog(deployDir);
	if (undeployed.size() > 0)
	__log.info("Successfully undeployed " + pkg);
	} finally {
	if (__log.isDebugEnabled()) {
	__log.debug("Trying to release the lock for " + duName);
	}
	unlock(duName);
	}
	}
	}
	}

	checkSystemCronConfigWatchDog(_systemCronConfigWatchDog);
	}

	@SuppressWarnings("unchecked")
	protected WatchDog ensureDeployXmlWatchDog(File deployFolder, File deployXml) {
	WatchDog ddWatchDog = dDWatchDogsByPath.get(deployXml.getAbsolutePath());
	if( ddWatchDog == null ) {
	ddWatchDog = WatchDog.watchFile(deployXml, new DDWatchDogObserver(deployFolder.getName()));
	dDWatchDogsByPath.put(deployXml.getAbsolutePath(), ddWatchDog);
	}

	return ddWatchDog;
	}

	@SuppressWarnings("unchecked")
	protected void checkDeployXmlWatchDog(WatchDog ddWatchDog) {
	ddWatchDog.check();
	}

	protected void disposeDeployXmlWatchDog(File deployDir) {
	dDWatchDogsByPath.remove(new File(deployDir, "deploy.xml").getAbsolutePath());
	}

	protected SystemSchedulesConfig createSystemSchedulesConfig(File configRoot) {
	return new SystemSchedulesConfig(configRoot);
	}

	@SuppressWarnings("unchecked")
	protected WatchDog createSystemCronConfigWatchDog(final CronScheduler cronScheduler) {
	return WatchDog.watchFile(_systemSchedulesConf.getSchedulesFile(),
	new WatchDog.DefaultObserver() {
	public void init() {
	cronScheduler.refreshSystemCronJobs(_systemSchedulesConf);
	}
	});
	}

	@SuppressWarnings("unchecked")
	protected void checkSystemCronConfigWatchDog(WatchDog ddWatchDog) {
	ddWatchDog.check();
	}

	/**
	* Thread that does the actual polling for new files.
	*/
	private class PollingThread extends Thread {
	private boolean _active = true;

	public PollingThread() {
	setName("DeploymentPoller");
	}

	/** Stop this poller, and block until it terminates. */
	void kill() {
	synchronized (this) {
	_active = false;
	this.notifyAll();
	}
	try {
	join();
	} catch (InterruptedException ie) {
	__log.error("Thread unexpectedly interrupted.", ie);
	}
	}

	public void run() {
	try {
	while (_active) {
	if (!_onHold)
	check();
	synchronized (this) {
	try {
	this.wait(POLL_TIME);
	} catch (InterruptedException e) {
	}
	}
	}
	} catch (Throwable t) {
	__log.error("Encountered an unexpected error. Exiting poller...", t);
	}
	}
	}

	public void hold() {
	_onHold = true;
	}

	public void release() {
	_onHold = false;
	}

	public void markAsDeployed(File file) {
	File deployedMarker = new File(_deployDir, file.getName() + ".deployed");
	try {
	boolean isCreated = deployedMarker.createNewFile();
	if (!isCreated) {
	__log.error("Error while creating file " + file.getName()
	+ ".deployed ,deployment could be inconsistent");
	}
	} catch (IOException e) {
	__log.error("Couldn't create marker file for " + file.getName());
	}
	}

	public void markAsUndeployed(File file) {
	File deployedMarker = new File(_deployDir, file.getName() + ".deployed");
	boolean isDeleted = deployedMarker.delete();
	if (!isDeleted) {
	__log
	.error("Error while deleting file "
	+ file.getName()
	+ ".deployed , please check if file is locked or if it really exist");
	}
	}

	@SuppressWarnings("unchecked")
	protected class DDWatchDogObserver extends WatchDog.DefaultObserver {
	private String deploymentPakage;

	public DDWatchDogObserver(String deploymentPakage) {
	this.deploymentPakage = deploymentPakage;
	}

	public void init() {
	_odeServer.getProcessStore().refreshSchedules(deploymentPakage);
	}
	}

	/**
	* Use to acquire the lock by poller
	*/
	private boolean pollerTryLock(String key) {
	if(clusterEnabled) {
	ClusterLock clusterLock = _odeServer.getBpelServer().getContexts().clusterManager.getDeploymentLock();
	clusterLock.putIfAbsent(key,key);
	return clusterLock.tryLock(key);
	}
	else return true;
	}

	private boolean unlock(String key) {
	if (clusterEnabled) {
	_odeServer.getBpelServer().getContexts().clusterManager.getDeploymentLock().unlock(key);
	}
	return true;
	}
	}