| // |
| // ======================================================================== |
| // Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd. |
| // ------------------------------------------------------------------------ |
| // All rights reserved. This program and the accompanying materials |
| // are made available under the terms of the Eclipse Public License v1.0 |
| // and Apache License v2.0 which accompanies this distribution. |
| // |
| // The Eclipse Public License is available at |
| // http://www.eclipse.org/legal/epl-v10.html |
| // |
| // The Apache License v2.0 is available at |
| // http://www.opensource.org/licenses/apache2.0.php |
| // |
| // You may elect to redistribute this code under either of these licenses. |
| // ======================================================================== |
| // |
| |
| package org.eclipse.jetty.gcloud.session; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.IOException; |
| import java.io.ObjectOutputStream; |
| import java.util.HashSet; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.concurrent.locks.ReentrantLock; |
| |
| import javax.servlet.http.HttpServletRequest; |
| |
| import org.eclipse.jetty.server.handler.ContextHandler; |
| import org.eclipse.jetty.server.handler.ContextHandler.Context; |
| import org.eclipse.jetty.server.session.AbstractSession; |
| import org.eclipse.jetty.server.session.AbstractSessionManager; |
| import org.eclipse.jetty.server.session.MemSession; |
| import org.eclipse.jetty.util.ClassLoadingObjectInputStream; |
| import org.eclipse.jetty.util.log.Log; |
| import org.eclipse.jetty.util.log.Logger; |
| import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; |
| import org.eclipse.jetty.util.thread.Scheduler; |
| |
| import com.google.cloud.datastore.Blob; |
| import com.google.cloud.datastore.Datastore; |
| import com.google.cloud.datastore.DatastoreException; |
| import com.google.cloud.datastore.DatastoreOptions; |
| import com.google.cloud.datastore.Entity; |
| import com.google.cloud.datastore.GqlQuery; |
| import com.google.cloud.datastore.Key; |
| import com.google.cloud.datastore.KeyFactory; |
| import com.google.cloud.datastore.Query; |
| import com.google.cloud.datastore.Query.ResultType; |
| import com.google.cloud.datastore.QueryResults; |
| |
| |
| |
| /** |
| * GCloudSessionManager |
| * |
| * |
| */ |
| public class GCloudSessionManager extends AbstractSessionManager |
| { |
| private final static Logger LOG = Log.getLogger("org.eclipse.jetty.server.session"); |
| |
| |
| public static final String KIND = "GCloudSession"; |
| public static final int DEFAULT_MAX_QUERY_RESULTS = 100; |
| public static final long DEFAULT_SCAVENGE_SEC = 600; |
| public static final int DEFAULT_BACKOFF_MS = 1000; //start at 1 sec |
| public static final int DEFAULT_MAX_RETRIES = 5; |
| |
| /** |
| * Sessions known to this node held in memory |
| */ |
| private ConcurrentHashMap<String, GCloudSessionManager.Session> _sessions; |
| |
| |
| /** |
| * The length of time a session can be in memory without being checked against |
| * the cluster. A value of 0 indicates that the session is never checked against |
| * the cluster - the current node is considered to be the master for the session. |
| * |
| */ |
| private long _staleIntervalSec = 0; |
| |
| protected Scheduler.Task _task; //scavenge task |
| protected Scheduler _scheduler; |
| protected Scavenger _scavenger; |
| protected long _scavengeIntervalMs = 1000L * DEFAULT_SCAVENGE_SEC; //10mins |
| protected boolean _ownScheduler; |
| |
| private Datastore _datastore; |
| private KeyFactory _keyFactory; |
| |
| |
| private SessionEntityConverter _converter; |
| |
| |
| private int _maxResults = DEFAULT_MAX_QUERY_RESULTS; |
| private int _backoffMs = DEFAULT_BACKOFF_MS; |
| private int _maxRetries = DEFAULT_MAX_RETRIES; |
| |
| |
| private boolean _dsSet; |
| |
| |
| /** |
| * Scavenger |
| * |
| */ |
| protected class Scavenger implements Runnable |
| { |
| |
| @Override |
| public void run() |
| { |
| try |
| { |
| scavenge(); |
| } |
| finally |
| { |
| if (_scheduler != null && _scheduler.isRunning()) |
| _task = _scheduler.schedule(this, _scavengeIntervalMs, TimeUnit.MILLISECONDS); |
| } |
| } |
| } |
| |
| /** |
| * SessionEntityConverter |
| * |
| * |
| */ |
| public class SessionEntityConverter |
| { |
| public final String CLUSTERID = "clusterId"; |
| public final String CONTEXTPATH = "contextPath"; |
| public final String VHOST = "vhost"; |
| public final String ACCESSED = "accessed"; |
| public final String LASTACCESSED = "lastAccessed"; |
| public final String CREATETIME = "createTime"; |
| public final String COOKIESETTIME = "cookieSetTime"; |
| public final String LASTNODE = "lastNode"; |
| public final String EXPIRY = "expiry"; |
| public final String MAXINACTIVE = "maxInactive"; |
| public final String ATTRIBUTES = "attributes"; |
| |
| |
| |
| public Entity entityFromSession (Session session, Key key) throws Exception |
| { |
| if (session == null) |
| return null; |
| |
| Entity entity = null; |
| |
| //serialize the attribute map |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| ObjectOutputStream oos = new ObjectOutputStream(baos); |
| oos.writeObject(session.getAttributeMap()); |
| oos.flush(); |
| |
| //turn a session into an entity |
| entity = Entity.builder(key) |
| .set(CLUSTERID, session.getId()) |
| .set(CONTEXTPATH, session.getContextPath()) |
| .set(VHOST, session.getVHost()) |
| .set(ACCESSED, session.getAccessed()) |
| .set(LASTACCESSED, session.getLastAccessedTime()) |
| .set(CREATETIME, session.getCreationTime()) |
| .set(COOKIESETTIME, session.getCookieSetTime()) |
| .set(LASTNODE,session.getLastNode()) |
| .set(EXPIRY, session.getExpiry()) |
| .set(MAXINACTIVE, session.getMaxInactiveInterval()) |
| .set(ATTRIBUTES, Blob.copyFrom(baos.toByteArray())).build(); |
| |
| return entity; |
| } |
| |
| public Session sessionFromEntity (Entity entity) throws Exception |
| { |
| if (entity == null) |
| return null; |
| |
| final AtomicReference<Session> reference = new AtomicReference<Session>(); |
| final AtomicReference<Exception> exception = new AtomicReference<Exception>(); |
| Runnable load = new Runnable() |
| { |
| public void run () |
| { |
| try |
| { |
| //turn an entity into a Session |
| String clusterId = entity.getString(CLUSTERID); |
| String contextPath = entity.getString(CONTEXTPATH); |
| String vhost = entity.getString(VHOST); |
| long accessed = entity.getLong(ACCESSED); |
| long lastAccessed = entity.getLong(LASTACCESSED); |
| long createTime = entity.getLong(CREATETIME); |
| long cookieSetTime = entity.getLong(COOKIESETTIME); |
| String lastNode = entity.getString(LASTNODE); |
| long expiry = entity.getLong(EXPIRY); |
| long maxInactive = entity.getLong(MAXINACTIVE); |
| Blob blob = (Blob) entity.getBlob(ATTRIBUTES); |
| |
| Session session = new Session (clusterId, createTime, accessed, maxInactive); |
| session.setLastNode(lastNode); |
| session.setContextPath(contextPath); |
| session.setVHost(vhost); |
| session.setCookieSetTime(cookieSetTime); |
| session.setLastAccessedTime(lastAccessed); |
| session.setLastNode(lastNode); |
| session.setExpiry(expiry); |
| try (ClassLoadingObjectInputStream ois = new ClassLoadingObjectInputStream(blob.asInputStream())) |
| { |
| Object o = ois.readObject(); |
| session.addAttributes((Map<String,Object>)o); |
| } |
| reference.set(session); |
| } |
| catch (Exception e) |
| { |
| exception.set(e); |
| } |
| } |
| }; |
| |
| if (_context==null) |
| load.run(); |
| else |
| _context.getContextHandler().handle(null,load); |
| |
| |
| if (exception.get() != null) |
| { |
| exception.get().printStackTrace(); |
| throw exception.get(); |
| } |
| |
| return reference.get(); |
| } |
| } |
| |
| |
| |
| |
| |
| /** |
| * Session |
| * |
| * Representation of a session in local memory. |
| */ |
| public class Session extends MemSession |
| { |
| |
| private ReentrantLock _lock = new ReentrantLock(); |
| |
| /** |
| * The (canonical) context path for with which this session is associated |
| */ |
| private String _contextPath; |
| |
| |
| |
| /** |
| * The time in msec since the epoch at which this session should expire |
| */ |
| private long _expiryTime; |
| |
| |
| /** |
| * Time in msec since the epoch at which this session was last read from cluster |
| */ |
| private long _lastSyncTime; |
| |
| |
| /** |
| * The workername of last node known to be managing the session |
| */ |
| private String _lastNode; |
| |
| |
| /** |
| * If dirty, session needs to be (re)sent to cluster |
| */ |
| protected boolean _dirty=false; |
| |
| |
| |
| |
| /** |
| * Any virtual hosts for the context with which this session is associated |
| */ |
| private String _vhost; |
| |
| |
| /** |
| * Count of how many threads are active in this session |
| */ |
| private AtomicInteger _activeThreads = new AtomicInteger(0); |
| |
| |
| |
| |
| /** |
| * A new session. |
| * |
| * @param request |
| */ |
| protected Session (HttpServletRequest request) |
| { |
| super(GCloudSessionManager.this,request); |
| long maxInterval = getMaxInactiveInterval(); |
| _expiryTime = (maxInterval <= 0 ? 0 : (System.currentTimeMillis() + maxInterval*1000L)); |
| _lastNode = getSessionIdManager().getWorkerName(); |
| setVHost(GCloudSessionManager.getVirtualHost(_context)); |
| setContextPath(GCloudSessionManager.getContextPath(_context)); |
| _activeThreads.incrementAndGet(); //access will not be called on a freshly created session so increment here |
| } |
| |
| |
| |
| |
| /** |
| * A restored session. |
| * |
| * @param sessionId |
| * @param created |
| * @param accessed |
| * @param maxInterval |
| */ |
| protected Session (String sessionId, long created, long accessed, long maxInterval) |
| { |
| super(GCloudSessionManager.this, created, accessed, sessionId); |
| _expiryTime = (maxInterval <= 0 ? 0 : (System.currentTimeMillis() + maxInterval*1000L)); |
| } |
| |
| /** |
| * Called on entry to the session. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSession#access(long) |
| */ |
| @Override |
| protected boolean access(long time) |
| { |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Access session({}) for context {} on worker {}", getId(), getContextPath(), getSessionIdManager().getWorkerName()); |
| try |
| { |
| |
| long now = System.currentTimeMillis(); |
| //lock so that no other thread can call access or complete until the first one has refreshed the session object if necessary |
| _lock.lock(); |
| //a request thread is entering |
| if (_activeThreads.incrementAndGet() == 1) |
| { |
| //if the first thread, check that the session in memory is not stale, if we're checking for stale sessions |
| if (getStaleIntervalSec() > 0 && (now - getLastSyncTime()) >= (getStaleIntervalSec() * 1000L)) |
| { |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Acess session({}) for context {} on worker {} stale session. Reloading.", getId(), getContextPath(), getSessionIdManager().getWorkerName()); |
| refresh(); |
| } |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn(e); |
| } |
| finally |
| { |
| _lock.unlock(); |
| } |
| |
| if (super.access(time)) |
| { |
| int maxInterval=getMaxInactiveInterval(); |
| _expiryTime = (maxInterval <= 0 ? 0 : (time + maxInterval*1000L)); |
| return true; |
| } |
| return false; |
| } |
| |
| |
| /** |
| * Exit from session |
| * @see org.eclipse.jetty.server.session.AbstractSession#complete() |
| */ |
| @Override |
| protected void complete() |
| { |
| super.complete(); |
| |
| //lock so that no other thread that might be calling access can proceed until this complete is done |
| _lock.lock(); |
| |
| try |
| { |
| //if this is the last request thread to be in the session |
| if (_activeThreads.decrementAndGet() == 0) |
| { |
| try |
| { |
| //an invalid session will already have been removed from the |
| //local session map and deleted from the cluster. If its valid save |
| //it to the cluster. |
| //TODO consider doing only periodic saves if only the last access |
| //time to the session changes |
| if (isValid()) |
| { |
| //if session still valid && its dirty or stale or never been synced, write it to the cluster |
| //otherwise, we just keep the updated last access time in memory |
| if (_dirty || getLastSyncTime() == 0 || isStale(System.currentTimeMillis())) |
| { |
| willPassivate(); |
| save(this); |
| didActivate(); |
| } |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn("Problem saving session({})",getId(), e); |
| } |
| finally |
| { |
| _dirty = false; |
| } |
| } |
| } |
| finally |
| { |
| _lock.unlock(); |
| } |
| } |
| |
| /** Test if the session is stale |
| * @param atTime |
| * @return true if the session is stale at the time given |
| */ |
| protected boolean isStale (long atTime) |
| { |
| return (getStaleIntervalSec() > 0) && (atTime - getLastSyncTime() >= (getStaleIntervalSec()*1000L)); |
| } |
| |
| |
| /** Test if the session is dirty |
| * @return true if the dirty flag is set |
| */ |
| protected boolean isDirty () |
| { |
| return _dirty; |
| } |
| |
| /** |
| * Expire the session. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSession#timeout() |
| */ |
| @Override |
| protected void timeout() |
| { |
| if (LOG.isDebugEnabled()) LOG.debug("Timing out session {}", getId()); |
| super.timeout(); |
| } |
| |
| |
| |
| /** |
| * Reload the session from the cluster. If the node that |
| * last managed the session from the cluster is ourself, |
| * then the session does not need refreshing. |
| * NOTE: this method MUST be called with sufficient locks |
| * in place to prevent 2 or more concurrent threads from |
| * simultaneously updating the session. |
| */ |
| private void refresh () |
| throws Exception |
| { |
| //get fresh copy from the cluster |
| Session fresh = load(makeKey(getClusterId(), _context)); |
| |
| //if the session no longer exists, invalidate |
| if (fresh == null) |
| { |
| invalidate(); |
| return; |
| } |
| |
| //cluster copy assumed to be the same as we were the last |
| //node to manage it |
| if (fresh.getLastNode().equals(getLastNode())) |
| return; |
| |
| setLastNode(getSessionIdManager().getWorkerName()); |
| |
| //prepare for refresh |
| willPassivate(); |
| |
| //if fresh has no attributes, remove them |
| if (fresh.getAttributes() == 0) |
| this.clearAttributes(); |
| else |
| { |
| //reconcile attributes |
| for (String key:fresh.getAttributeMap().keySet()) |
| { |
| Object freshvalue = fresh.getAttribute(key); |
| |
| //session does not already contain this attribute, so bind it |
| if (getAttribute(key) == null) |
| { |
| doPutOrRemove(key,freshvalue); |
| bindValue(key,freshvalue); |
| } |
| else //session already contains this attribute, update its value |
| { |
| doPutOrRemove(key,freshvalue); |
| } |
| |
| } |
| // cleanup, remove values from session, that don't exist in data anymore: |
| for (String key : getNames()) |
| { |
| if (fresh.getAttribute(key) == null) |
| { |
| Object oldvalue = getAttribute(key); |
| doPutOrRemove(key,null); |
| unbindValue(key,oldvalue); |
| } |
| } |
| } |
| //finish refresh |
| didActivate(); |
| } |
| |
| |
| public void setExpiry (long expiry) |
| { |
| _expiryTime = expiry; |
| } |
| |
| |
| public long getExpiry () |
| { |
| return _expiryTime; |
| } |
| |
| public boolean isExpiredAt (long time) |
| { |
| if (_expiryTime <= 0) |
| return false; //never expires |
| |
| return (_expiryTime <= time); |
| } |
| |
| public void swapId (String newId, String newNodeId) |
| { |
| //TODO probably synchronize rather than use the access/complete lock? |
| _lock.lock(); |
| setClusterId(newId); |
| setNodeId(newNodeId); |
| _lock.unlock(); |
| } |
| |
| @Override |
| public void setAttribute (String name, Object value) |
| { |
| Object old = changeAttribute(name, value); |
| if (value == null && old == null) |
| return; //if same as remove attribute but attribute was already removed, no change |
| |
| _dirty = true; |
| } |
| |
| |
| public String getContextPath() |
| { |
| return _contextPath; |
| } |
| |
| |
| public void setContextPath(String contextPath) |
| { |
| this._contextPath = contextPath; |
| } |
| |
| |
| public String getVHost() |
| { |
| return _vhost; |
| } |
| |
| |
| public void setVHost(String vhost) |
| { |
| this._vhost = vhost; |
| } |
| |
| public String getLastNode() |
| { |
| return _lastNode; |
| } |
| |
| |
| public void setLastNode(String lastNode) |
| { |
| _lastNode = lastNode; |
| } |
| |
| |
| public long getLastSyncTime() |
| { |
| return _lastSyncTime; |
| } |
| |
| |
| public void setLastSyncTime(long lastSyncTime) |
| { |
| _lastSyncTime = lastSyncTime; |
| } |
| |
| } |
| |
| |
| |
| |
| public void setDatastore (Datastore datastore) |
| { |
| _datastore = datastore; |
| _dsSet = true; |
| } |
| |
| |
| |
| /** |
| * Start the session manager. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#doStart() |
| */ |
| @Override |
| public void doStart() throws Exception |
| { |
| if (_sessionIdManager == null) |
| throw new IllegalStateException("No session id manager defined"); |
| |
| if (!_dsSet) |
| _datastore = DatastoreOptions.defaultInstance().service(); |
| _keyFactory = _datastore.newKeyFactory().kind(KIND); |
| _converter = new SessionEntityConverter(); |
| _sessions = new ConcurrentHashMap<String, Session>(); |
| |
| //try and use a common scheduler, fallback to own |
| _scheduler = getSessionHandler().getServer().getBean(Scheduler.class); |
| if (_scheduler == null) |
| { |
| _scheduler = new ScheduledExecutorScheduler(); |
| _ownScheduler = true; |
| _scheduler.start(); |
| } |
| else if (!_scheduler.isStarted()) |
| throw new IllegalStateException("Shared scheduler not started"); |
| |
| setScavengeIntervalSec(getScavengeIntervalSec()); |
| |
| super.doStart(); |
| } |
| |
| |
| /** |
| * Stop the session manager. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#doStop() |
| */ |
| @Override |
| public void doStop() throws Exception |
| { |
| super.doStop(); |
| |
| if (_task!=null) |
| _task.cancel(); |
| _task=null; |
| if (_ownScheduler && _scheduler !=null) |
| _scheduler.stop(); |
| _scheduler = null; |
| |
| _sessions.clear(); |
| _sessions = null; |
| |
| if (!_dsSet) |
| _datastore = null; |
| } |
| |
| |
| |
| /** |
| * Look for sessions in local memory that have expired. |
| */ |
| public void scavenge () |
| { |
| try |
| { |
| //scavenge in the database every so often |
| scavengeGCloudDataStore(); |
| } |
| catch (Exception e) |
| { |
| LOG.warn("Problem scavenging", e); |
| } |
| } |
| |
| |
| |
| protected void scavengeGCloudDataStore() |
| throws Exception |
| { |
| |
| //query the datastore for sessions that have expired |
| long now = System.currentTimeMillis(); |
| |
| //give a bit of leeway so we don't immediately something that has only just expired a nanosecond ago |
| now = now - (_scavengeIntervalMs/2); |
| |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Scavenging for sessions expired before "+now); |
| |
| |
| GqlQuery.Builder builder = Query.gqlQueryBuilder(ResultType.ENTITY, "select * from "+KIND+" where expiry < @1 limit "+_maxResults); |
| builder.allowLiteral(true); |
| builder.addBinding(now); |
| Query<Entity> query = builder.build(); |
| QueryResults<Entity> results = _datastore.run(query); |
| |
| while (results.hasNext()) |
| { |
| Entity sessionEntity = results.next(); |
| scavengeSession(sessionEntity); |
| } |
| |
| } |
| |
| /** |
| * Scavenge a session that has expired |
| * @param e the session info from datastore |
| * @throws Exception |
| */ |
| protected void scavengeSession (Entity e) |
| throws Exception |
| { |
| long now = System.currentTimeMillis(); |
| Session session = _converter.sessionFromEntity(e); |
| if (session == null) |
| return; |
| |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Scavenging session: {}",session.getId()); |
| //if the session isn't in memory already, put it there so we can do a normal timeout call |
| Session memSession = _sessions.putIfAbsent(session.getId(), session); |
| if (memSession == null) |
| { |
| memSession = session; |
| } |
| |
| //final check |
| if (memSession.isExpiredAt(now)) |
| { |
| if (LOG.isDebugEnabled()) LOG.debug("Session {} is definitely expired", memSession.getId()); |
| memSession.timeout(); |
| } |
| } |
| |
| public long getScavengeIntervalSec () |
| { |
| return _scavengeIntervalMs/1000; |
| } |
| |
| |
| |
| /** |
| * Set the interval between runs of the scavenger. It should not be run too |
| * often. |
| * |
| * |
| * @param sec the number of seconds between scavenge cycles |
| */ |
| public void setScavengeIntervalSec (long sec) |
| { |
| |
| long old_period=_scavengeIntervalMs; |
| long period=sec*1000L; |
| |
| _scavengeIntervalMs=period; |
| |
| if (_scavengeIntervalMs > 0) |
| { |
| //add a bit of variability into the scavenge time so that not all |
| //nodes with the same scavenge time sync up |
| long tenPercent = _scavengeIntervalMs/10; |
| if ((System.currentTimeMillis()%2) == 0) |
| _scavengeIntervalMs += tenPercent; |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Scavenging every "+_scavengeIntervalMs+" ms"); |
| } |
| else |
| { |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Scavenging disabled"); |
| } |
| |
| |
| |
| synchronized (this) |
| { |
| if (_scheduler != null && (period!=old_period || _task==null)) |
| { |
| //clean up any previously scheduled scavenger |
| if (_task!=null) |
| _task.cancel(); |
| |
| //start a new one |
| if (_scavengeIntervalMs > 0) |
| { |
| if (_scavenger == null) |
| _scavenger = new Scavenger(); |
| |
| _task = _scheduler.schedule(_scavenger,_scavengeIntervalMs,TimeUnit.MILLISECONDS); |
| } |
| } |
| } |
| } |
| |
| |
| public long getStaleIntervalSec() |
| { |
| return _staleIntervalSec; |
| } |
| |
| |
| public void setStaleIntervalSec(long staleIntervalSec) |
| { |
| _staleIntervalSec = staleIntervalSec; |
| } |
| |
| |
| public int getMaxResults() |
| { |
| return _maxResults; |
| } |
| |
| |
| public void setMaxResults(int maxResults) |
| { |
| if (_maxResults <= 0) |
| _maxResults = DEFAULT_MAX_QUERY_RESULTS; |
| else |
| _maxResults = maxResults; |
| } |
| |
| |
| /** |
| * Add a new session for the context related to this session manager |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#addSession(org.eclipse.jetty.server.session.AbstractSession) |
| */ |
| @Override |
| protected void addSession(AbstractSession session) |
| { |
| if (session==null) |
| return; |
| |
| if (LOG.isDebugEnabled()) LOG.debug("Adding session({}) to session manager for context {} on worker {}",session.getClusterId(), getContextPath(getContext()),getSessionIdManager().getWorkerName() + " with lastnode="+((Session)session).getLastNode()); |
| _sessions.put(session.getClusterId(), (Session)session); |
| |
| try |
| { |
| session.willPassivate(); |
| save(((GCloudSessionManager.Session)session)); |
| session.didActivate(); |
| |
| } |
| catch (Exception e) |
| { |
| LOG.warn("Unable to store new session id="+session.getId() , e); |
| } |
| } |
| |
| /** |
| * Ask the cluster for the session. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#getSession(java.lang.String) |
| */ |
| @Override |
| public AbstractSession getSession(String idInCluster) |
| { |
| Session session = null; |
| |
| //try and find the session in this node's memory |
| Session memSession = (Session)_sessions.get(idInCluster); |
| |
| if (LOG.isDebugEnabled()) |
| LOG.debug("getSession({}) {} in session map",idInCluster,(memSession==null?"not":"")); |
| |
| long now = System.currentTimeMillis(); |
| try |
| { |
| //if the session is not in this node's memory, then load it from the datastore |
| if (memSession == null) |
| { |
| if (LOG.isDebugEnabled()) |
| LOG.debug("getSession({}): loading session data from cluster", idInCluster); |
| |
| session = load(makeKey(idInCluster, _context)); |
| if (session != null) |
| { |
| //Check that it wasn't expired |
| if (session.getExpiry() > 0 && session.getExpiry() <= now) |
| { |
| if (LOG.isDebugEnabled()) LOG.debug("getSession ({}): Session expired", idInCluster); |
| //ensure that the session id for the expired session is deleted so that a new session with the |
| //same id cannot be created (because the idInUse() test would succeed) |
| ((GCloudSessionIdManager)getSessionIdManager()).removeSession(session); |
| return null; |
| } |
| |
| //Update the last worker node to me |
| session.setLastNode(getSessionIdManager().getWorkerName()); |
| //TODO consider saving session here if lastNode was not this node |
| |
| //Check that another thread hasn't loaded the same session |
| Session existingSession = _sessions.putIfAbsent(idInCluster, session); |
| if (existingSession != null) |
| { |
| //use the one that the other thread inserted |
| session = existingSession; |
| LOG.debug("getSession({}): using session loaded by another request thread ", idInCluster); |
| } |
| else |
| { |
| //indicate that the session was reinflated |
| session.didActivate(); |
| LOG.debug("getSession({}): loaded session from cluster", idInCluster); |
| } |
| return session; |
| } |
| else |
| { |
| //The requested session does not exist anywhere in the cluster |
| LOG.debug("getSession({}): No session in cluster matching",idInCluster); |
| return null; |
| } |
| } |
| else |
| { |
| //The session exists in this node's memory |
| LOG.debug("getSession({}): returning session from local memory ", memSession.getClusterId()); |
| return memSession; |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn("Unable to load session="+idInCluster, e); |
| return null; |
| } |
| } |
| |
| |
| |
| /** |
| * The session manager is stopping. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#shutdownSessions() |
| */ |
| @Override |
| protected void shutdownSessions() throws Exception |
| { |
| Set<String> keys = new HashSet<String>(_sessions.keySet()); |
| for (String key:keys) |
| { |
| Session session = _sessions.remove(key); //take the session out of the session list |
| //If the session is dirty, then write it to the cluster. |
| //If the session is simply stale do NOT write it to the cluster, as some other node |
| //may have started managing that session - this means that the last accessed/expiry time |
| //will not be updated, meaning it may look like it can expire sooner than it should. |
| try |
| { |
| if (session.isDirty()) |
| { |
| if (LOG.isDebugEnabled()) |
| LOG.debug("Saving dirty session {} before exiting ", session.getId()); |
| save(session); |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn(e); |
| } |
| } |
| } |
| |
| |
| @Override |
| protected AbstractSession newSession(HttpServletRequest request) |
| { |
| return new Session(request); |
| } |
| |
| /** |
| * Remove a session from local memory, and delete it from |
| * the cluster cache. |
| * |
| * @see org.eclipse.jetty.server.session.AbstractSessionManager#removeSession(java.lang.String) |
| */ |
| @Override |
| protected boolean removeSession(String idInCluster) |
| { |
| Session session = (Session)_sessions.remove(idInCluster); |
| try |
| { |
| if (session != null) |
| { |
| delete(session); |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn("Problem deleting session id="+idInCluster, e); |
| } |
| return session!=null; |
| } |
| |
| |
| |
| |
| @Override |
| public void renewSessionId(String oldClusterId, String oldNodeId, String newClusterId, String newNodeId) |
| { |
| Session session = null; |
| try |
| { |
| //take the session with that id out of our managed list |
| session = (Session)_sessions.remove(oldClusterId); |
| if (session != null) |
| { |
| //TODO consider transactionality and ramifications if the session is live on another node |
| delete(session); //delete the old session from the cluster |
| session.swapId(newClusterId, newNodeId); //update the session |
| _sessions.put(newClusterId, session); //put it into managed list under new key |
| save(session); //put the session under the new id into the cluster |
| } |
| } |
| catch (Exception e) |
| { |
| LOG.warn(e); |
| } |
| |
| super.renewSessionId(oldClusterId, oldNodeId, newClusterId, newNodeId); |
| } |
| |
| |
| /** |
| * Load a session from the clustered cache. |
| * |
| * @param key the unique datastore key for the session |
| * @return the Session object restored from datastore |
| */ |
| protected Session load (Key key) |
| throws Exception |
| { |
| if (_datastore == null) |
| throw new IllegalStateException("No DataStore"); |
| |
| if (LOG.isDebugEnabled()) LOG.debug("Loading session {} from DataStore ", key); |
| |
| Entity entity = _datastore.get(key); |
| if (entity == null) |
| { |
| if (LOG.isDebugEnabled()) LOG.debug("No session {} in DataStore ",key); |
| return null; |
| } |
| else |
| { |
| Session session = _converter.sessionFromEntity(entity); |
| session.setLastSyncTime(System.currentTimeMillis()); |
| return session; |
| } |
| } |
| |
| |
| |
| /** |
| * Save or update the session to the cluster cache |
| * |
| * @param session the session to save to datastore |
| * @throws Exception |
| */ |
| protected void save (GCloudSessionManager.Session session) |
| throws Exception |
| { |
| if (_datastore == null) |
| throw new IllegalStateException("No DataStore"); |
| |
| if (LOG.isDebugEnabled()) LOG.debug("Writing session {} to DataStore", session.getId()); |
| |
| Entity entity = _converter.entityFromSession(session, makeKey(session, _context)); |
| |
| //attempt the update with exponential back-off |
| int backoff = getBackoffMs(); |
| int attempts; |
| for (attempts = 0; attempts < getMaxRetries(); attempts++) |
| { |
| try |
| { |
| _datastore.put(entity); |
| session.setLastSyncTime(System.currentTimeMillis()); |
| return; |
| } |
| catch (DatastoreException e) |
| { |
| if (e.retryable()) |
| { |
| if (LOG.isDebugEnabled()) LOG.debug("Datastore put retry {} waiting {}ms", attempts, backoff); |
| |
| try |
| { |
| Thread.currentThread().sleep(backoff); |
| } |
| catch (InterruptedException x) |
| { |
| } |
| backoff *= 2; |
| } |
| else |
| { |
| throw e; |
| } |
| } |
| } |
| throw new IOException("Retries exhausted"); |
| } |
| |
| |
| |
| public int getMaxRetries() |
| { |
| return _maxRetries; |
| } |
| |
| |
| public int getBackoffMs() |
| { |
| return _backoffMs; |
| } |
| |
| |
| /** |
| * @param backoffMs the backoffMs to set |
| */ |
| public void setBackoffMs(int backoffMs) |
| { |
| _backoffMs = backoffMs; |
| } |
| |
| |
| /** |
| * @param maxRetries the maxRetries to set |
| */ |
| public void setMaxRetries(int maxRetries) |
| { |
| _maxRetries = maxRetries; |
| } |
| |
| |
| /** |
| * Remove the session from the cluster cache. |
| * |
| * @param session the session to delete from datastore |
| */ |
| protected void delete (GCloudSessionManager.Session session) |
| { |
| if (_datastore == null) |
| throw new IllegalStateException("No DataStore"); |
| if (LOG.isDebugEnabled()) LOG.debug("Removing session {} from DataStore", session.getId()); |
| _datastore.delete(makeKey(session, _context)); |
| } |
| |
| |
| /** |
| * Invalidate a session for this context with the given id |
| * |
| * @param idInCluster the id of the session to invalidate |
| */ |
| public void invalidateSession (String idInCluster) |
| { |
| Session session = (Session)_sessions.get(idInCluster); |
| |
| if (session != null) |
| { |
| session.invalidate(); |
| } |
| } |
| |
| |
| /** |
| * Make a unique key for this session. |
| * As the same session id can be used across multiple contexts, to |
| * make it unique, the key must be composed of: |
| * <ol> |
| * <li>the id</li> |
| * <li>the context path</li> |
| * <li>the virtual hosts</li> |
| * </ol> |
| * |
| * @param session the session for which the key should be created |
| * @param context the context to which the session belongs |
| * @return a unique datastore key for the session |
| */ |
| protected Key makeKey (Session session, Context context) |
| { |
| return makeKey(session.getId(), context); |
| } |
| |
| /** |
| * Make a unique key for this session. |
| * As the same session id can be used across multiple contexts, to |
| * make it unique, the key must be composed of: |
| * <ol> |
| * <li>the id</li> |
| * <li>the context path</li> |
| * <li>the virtual hosts</li> |
| * </ol> |
| * |
| * @param id the id of the session for which the key should be created |
| * @param context the context to which the session belongs |
| * @return a unique datastore key for the session |
| */ |
| protected Key makeKey (String id, Context context) |
| { |
| return _keyFactory.newKey(canonicalizeKey(id,context)); |
| } |
| |
| |
| /** |
| * Make a unique string from the session id and info from its Context |
| * @param id the id of the Session |
| * @param context the Context in which the Session exists |
| * @return a unique string representing the id of the session in the context |
| */ |
| protected String canonicalizeKey(String id, Context context) |
| { |
| String key = getContextPath(context); |
| key = key + "_" + getVirtualHost(context); |
| key = key+"_"+id; |
| return key; |
| } |
| |
| /** |
| * Turn the context path into an acceptable string |
| * |
| * @param context a context |
| * @return a stringified version of the context |
| */ |
| private static String getContextPath (ContextHandler.Context context) |
| { |
| return canonicalize (context.getContextPath()); |
| } |
| |
| /** |
| * Get the first virtual host for the context. |
| * |
| * Used to help identify the exact session/contextPath. |
| * |
| * @param context a context |
| * @return a stringified form of the virtual hosts for the context, 0.0.0.0 if none are defined |
| */ |
| private static String getVirtualHost (ContextHandler.Context context) |
| { |
| String vhost = "0.0.0.0"; |
| |
| if (context==null) |
| return vhost; |
| |
| String [] vhosts = context.getContextHandler().getVirtualHosts(); |
| if (vhosts==null || vhosts.length==0 || vhosts[0]==null) |
| return vhost; |
| |
| return vhosts[0]; |
| } |
| |
| /** |
| * Make an acceptable name from a context path. |
| * |
| * @param path a context path |
| * @return a stringified form of the context path |
| */ |
| private static String canonicalize (String path) |
| { |
| if (path==null) |
| return ""; |
| |
| return path.replace('/', '_').replace('.','_').replace('\\','_'); |
| } |
| |
| } |
