Merge branch 'master' into permission-vertical

Change-Id: I6111392e33d082fbc6d0bb5802b8776353ef5bcd
This commit is contained in:
Axel Uhl committed 2019-02-05 00:27:36 +01:00
commit 619e75fa17
15 files changed
+347 -87

No files matched your search

@@ -139,9 +139,9 @@ public class SimpleGridFileReadingTest {
// alternatively for the 4-6 day forecast: http://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.oceanic/VP.004-007/
// validate graphically at http://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.oceanic/VP.004-007/
final Formatter errorLog = new Formatter(System.err);
Files.copy(new URL("http://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.conus/VP.001-003/ds.wdir.bin").openStream(),
Files.copy(new URL("https://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.conus/VP.001-003/ds.wdir.bin").openStream(),
new File("resources/ds.wdir.bin").toPath(), StandardCopyOption.REPLACE_EXISTING);
Files.copy(new URL("http://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.conus/VP.001-003/ds.wspd.bin").openStream(),
Files.copy(new URL("https://tgftp.nws.noaa.gov/SL.us008001/ST.expr/DF.gr2/DC.ndfd/AR.conus/VP.001-003/ds.wspd.bin").openStream(),
new File("resources/ds.wspd.bin").toPath(), StandardCopyOption.REPLACE_EXISTING);
FeatureDataset dataSetDirection = FeatureDatasetFactoryManager.open(FeatureType.ANY, "resources/ds.wdir.bin", /* task */ null, errorLog);
@@ -18,6 +18,7 @@ import com.sap.sse.InvalidDateException;
import com.sap.sse.common.TimePoint;
import com.sap.sse.common.impl.MillisecondsTimePoint;
import com.sap.sse.security.SecurityService;
import com.sap.sse.replication.ReplicationService;
import com.sap.sse.util.DateParser;
import com.sun.jersey.api.core.ResourceContext;
@@ -56,6 +57,12 @@ public abstract class AbstractSailingServerResource {
return tracker.getService();
}
public ReplicationService getReplicationService() {
@SuppressWarnings("unchecked")
ServiceTracker<ReplicationService, ReplicationService> tracker = (ServiceTracker<ReplicationService, ReplicationService>) servletContext.getAttribute(RestServletContainer.REPLICATION_SERVICE_TRACKER_NAME);
return tracker.getService();
}
protected Regatta findRegattaByName(String regattaName) {
Regatta regatta = getService().getRegattaByName(regattaName);
if (regatta == null && regattaName.contains(SLASH_ENCODING)) {
@@ -10,6 +10,7 @@ import org.osgi.util.tracker.ServiceTracker;
import com.sap.sailing.server.interfaces.RacingEventService;
import com.sap.sse.datamining.DataMiningServer;
import com.sap.sse.security.SecurityService;
import com.sap.sse.replication.ReplicationService;
import com.sun.jersey.spi.container.servlet.ServletContainer;
public class RestServletContainer extends ServletContainer {
@@ -19,12 +20,16 @@ public class RestServletContainer extends ServletContainer {
public static final String RACING_EVENT_SERVICE_TRACKER_NAME = "racingEventServiceTracker";
public static final String REPLICATION_SERVICE_TRACKER_NAME = "replicationServiceTracker";
public static final String DATA_MINING_SERVER_TRACKER_NAME = "dataMiningServerTracker";
public static final String SECURITY_SERVICE_TRACKER_NAME = "securityServiceTracker";
private ServiceTracker<RacingEventService, RacingEventService> racingEventServiceTracker;
private ServiceTracker<ReplicationService, ReplicationService> replicationServiceTracker;
private ServiceTracker<DataMiningServer, DataMiningServer> dataMiningServerTracker;
private ServiceTracker<SecurityService, SecurityService> securityServiceTracker;
@@ -47,11 +52,14 @@ public class RestServletContainer extends ServletContainer {
BundleContext context = (BundleContext) config.getServletContext().getAttribute(OSGI_RFC66_WEBBUNDLE_BUNDLECONTEXT_NAME);
racingEventServiceTracker = new ServiceTracker<RacingEventService, RacingEventService>(context, RacingEventService.class.getName(), null);
racingEventServiceTracker.open();
replicationServiceTracker = new ServiceTracker<ReplicationService, ReplicationService>(context, ReplicationService.class.getName(), null);
replicationServiceTracker.open();
dataMiningServerTracker = new ServiceTracker<DataMiningServer, DataMiningServer>(context, DataMiningServer.class, null);
dataMiningServerTracker.open();
securityServiceTracker = new ServiceTracker<SecurityService, SecurityService>(context, SecurityService.class, null);
securityServiceTracker.open();
config.getServletContext().setAttribute(RACING_EVENT_SERVICE_TRACKER_NAME, racingEventServiceTracker);
config.getServletContext().setAttribute(REPLICATION_SERVICE_TRACKER_NAME, replicationServiceTracker);
config.getServletContext().setAttribute(DATA_MINING_SERVER_TRACKER_NAME, dataMiningServerTracker);
config.getServletContext().setAttribute(SECURITY_SERVICE_TRACKER_NAME, securityServiceTracker);
}
@@ -1,5 +1,6 @@
package com.sap.sailing.server.gateway.jaxrs.api;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
@@ -10,6 +11,8 @@ import org.json.simple.JSONObject;
import com.sap.sailing.server.gateway.jaxrs.AbstractSailingServerResource;
import com.sap.sailing.server.interfaces.RacingEventService;
import com.sap.sse.replication.ReplicationService;
import com.sap.sse.replication.ReplicationStatus;
@Path("/v1/status")
public class StatusResource extends AbstractSailingServerResource {
@@ -20,10 +23,15 @@ public class StatusResource extends AbstractSailingServerResource {
final RacingEventService service = getService();
result.put("numberofracestorestore", service.getNumberOfTrackedRacesToRestore());
result.put("numberofracesrestored", service.getNumberOfTrackedRacesRestored());
result.put("isreplica", service.getMasterDescriptor() != null);
result.put("initialloadrunning", service.isCurrentlyFillingFromInitialLoad());
result.put("available", service.getNumberOfTrackedRacesRestored() >= service.getNumberOfTrackedRacesToRestore() &&
(service.getMasterDescriptor() == null || !service.isCurrentlyFillingFromInitialLoad()));
return Response.ok(result.toJSONString()).header("Content-Type", MediaType.APPLICATION_JSON + ";charset=UTF-8").build();
final ReplicationService replicationService = getReplicationService();
final ReplicationStatus replicationStatus = replicationService == null ? null : replicationService.getStatus();
if (replicationStatus != null) {
result.put("replication", replicationStatus.toJSONObject());
}
final boolean available = service.getNumberOfTrackedRacesRestored() >= service.getNumberOfTrackedRacesToRestore() &&
(replicationStatus == null || replicationStatus.isAvailable());
result.put("available", available);
return Response.status(available ? HttpServletResponse.SC_OK : HttpServletResponse.SC_SERVICE_UNAVAILABLE).
entity(result.toJSONString()).header("Content-Type", MediaType.APPLICATION_JSON + ";charset=UTF-8").build();
}
}
@@ -18,6 +18,8 @@ Import-Package: com.sap.sse.security,
javax.servlet.descriptor;version="3.1.0",
javax.servlet.http;version="3.1.0",
org.apache.commons.io,
org.apache.shiro;version="1.2.2",
org.apache.shiro.subject;version="1.2.2",
org.json.simple,
org.json.simple.parser,
org.osgi.framework,
@@ -10,6 +10,8 @@ import javax.servlet.ServletContext;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.shiro.SecurityUtils;
import org.apache.shiro.subject.Subject;
import com.google.gwt.user.client.rpc.SerializationException;
import com.google.gwt.user.server.rpc.RPCRequest;
@@ -136,7 +138,16 @@ public abstract class ProxiedRemoteServiceServlet extends RemoteServiceServlet {
final Duration processingDuration = startOfProcessing.until(endOfProcessing);
final Duration serializationDuration = endOfProcessing.until(afterResultSerialization);
final Duration totalDuration = startOfProcessing.until(afterResultSerialization);
logger.log(Level.WARNING, "GWT RPC Request "+request.getMethod()+" with parameters "+Arrays.toString(request.getParameters())+" on "+this+
final String username;
final Subject subject = SecurityUtils.getSubject();
if (subject != null) {
username = subject.getPrincipal() == null ? null : subject.getPrincipal().toString();
} else {
username = null;
}
logger.log(Level.WARNING, "GWT RPC Request "+request.getMethod()+
" by user "+username+
" with parameters "+Arrays.toString(request.getParameters())+" on "+this+
" took "+processingDuration+" to process, "+serializationDuration+" to serialize the response, so "+
totalDuration+" in total.");
}
@@ -7,7 +7,8 @@ Bundle-Vendor: SAP
Automatic-Module-Name: com.sap.sse.replication.interfaces
Bundle-RequiredExecutionEnvironment: JavaSE-1.8
Export-Package: com.sap.sse.replication
Import-Package: com.rabbitmq.client;version="2.8.4"
Import-Package: com.rabbitmq.client;version="2.8.4",
org.json.simple
Require-Bundle: com.sap.sse.operationaltransformation;bundle-version="1.0.0",
com.sap.sse.common;bundle-version="1.0.0",
com.sap.sse;bundle-version="1.0.0"
@@ -121,4 +121,10 @@ public interface ReplicationService {
*
*/
boolean isReplicationStarting();
/**
* Provides a compound overview of this service and the {@link Replicable}s it manages.
* This can be used in compound server availability checks, for example.
*/
ReplicationStatus getStatus();
}
@@ -0,0 +1,44 @@
package com.sap.sse.replication;
import org.json.simple.JSONObject;
public interface ReplicationStatus {
boolean isReplica();
boolean isReplicationStarting();
boolean isSuspended();
boolean isStopped();
long getMessageQueueLength();
Iterable<String> getReplicableIdsAsStrings();
/**
* The queue length or {@code null} if no queue exists for the replicable with ID {@link code replicableIdAsString}
*/
Integer getOperationQueueLengthsByReplicableIdAsString(String replicableIdAsString);
int getTotalOperationQueueLength();
/**
* @return {@code true} if the initial load for the replicable with ID {@link code replicableIdAsString} is
* currently running, {@code false} if not, and {@code null} if the replicable with ID {@link code replicableIdAsString}
* is not known.
*/
Boolean isInitialLoadRunning(String replicableIdAsString);
/**
* An availability report based on the replication status of the overall service and all of the
* {@link Replicable}s managed by it. This method reports {@code true} if this server is not a replica,
* or in case it is a replica the replication is not currently starting and no initial load is currently
* running for any replicable managed by this service.<p>
*
* Note that if this server is not a replica there may still be other start-up processes going on, such as
* the restoring of races that need to be loaded.
*/
boolean isAvailable();
JSONObject toJSONObject();
}
@@ -123,50 +123,64 @@ public class Activator implements BundleActivator {
final OSGiReplicableTracker replicablesProvider = new OSGiReplicableTracker(bundleContext);
serverReplicationMasterService = new ReplicationServiceImpl(
exchangeName, exchangeHost, exchangePort, replicationInstancesManager, replicablesProvider);
String replicateOnStart = System.getProperty(PROPERTY_NAME_REPLICATE_ON_START);
final boolean autoReplicationRequested = replicateOnStart != null && !replicateOnStart.isEmpty();
if (autoReplicationRequested) {
// set before registering the service; clients discovering it therefore cannot accidentally
// discover it before this flag has been set.
serverReplicationMasterService.setReplicationStarting(true);
}
bundleContext.registerService(ReplicationService.class, serverReplicationMasterService, null);
logger.info("Registered replication service "+serverReplicationMasterService+" using exchange name "+exchangeName+" on host "+exchangeHost);
checkIfAutomaticReplicationShouldStart(serverReplicationMasterService, exchangeName, replicablesProvider);
if (autoReplicationRequested) {
triggerAutomaticReplication(serverReplicationMasterService, replicateOnStart, exchangeName, replicablesProvider);
}
}
private void checkIfAutomaticReplicationShouldStart(ReplicationService serverReplicationMasterService, String masterExchangeName, final ReplicablesProvider replicablesProvider) {
String replicateOnStart = System.getProperty(PROPERTY_NAME_REPLICATE_ON_START);
if (replicateOnStart != null && !replicateOnStart.isEmpty()) {
serverReplicationMasterService.setReplicationStarting(true);
final String[] replicableIdsAsStrings = replicateOnStart.split(",");
new Thread("ServiceTracker waiting for Replicables "+Arrays.toString(replicableIdsAsStrings)) {
@Override
public void run() {
logger.info("Waiting for Replicables " + Arrays.toString(replicableIdsAsStrings) +
" before firing up replication automatically...");
final List<Replicable<?, ?>> replicables = new ArrayList<>();
for (String replicableIdAsString : replicableIdsAsStrings) {
Replicable<?, ?> replicable = replicablesProvider.getReplicable(replicableIdAsString, /* wait */true);
logger.info("Obtained Replicable " + replicableIdAsString);
replicables.add(replicable);
}
logger.info("Configuration requested automatic replication for replicables "+
Arrays.toString(replicableIdsAsStrings)+". Starting it up...");
String replicateFromExchangeName = System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_EXCHANGE_NAME);
if (replicateFromExchangeName == null) {
replicateFromExchangeName = masterExchangeName;
}
ReplicationMasterDescriptorImpl master = new ReplicationMasterDescriptorImpl(
System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_QUEUE_HOST),
replicateFromExchangeName,
Integer.valueOf(System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_QUEUE_PORT).trim()),
serverReplicationMasterService.getServerIdentifier().toString(),
System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_SERVLET_HOST),
Integer.valueOf(System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_SERVLET_PORT).trim()), replicables);
try {
serverReplicationMasterService.startToReplicateFrom(master);
serverReplicationMasterService.setReplicationStarting(false);
logger.info("Automatic replication has been started.");
} catch (ClassNotFoundException | IOException | InterruptedException e) {
logger.log(Level.SEVERE, "Error with automatic replication from "+master, e);
}
/**
* Assumes that automatic replication was requested and that the {@code serverReplicationMasterService} has already been
* marked as {@link ReplicationService#setReplicationStarting(boolean) setReplicationStarting(true)}. When replication
* has been requested from all replicables, {@link ReplicationService#setReplicationStarting(boolean) setReplicationStarting(false)}
* will be called by this method to indicate that the start-up sequence for the service has completed; now clients may need
* to wait for the individual replicables to finish their initial load.
*/
private void triggerAutomaticReplication(ReplicationService serverReplicationMasterService, String replicateOnStart, String masterExchangeName, final ReplicablesProvider replicablesProvider) {
assert replicateOnStart != null && !replicateOnStart.isEmpty();
serverReplicationMasterService.setReplicationStarting(true);
final String[] replicableIdsAsStrings = replicateOnStart.split(",");
new Thread("ServiceTracker waiting for Replicables "+Arrays.toString(replicableIdsAsStrings)) {
@Override
public void run() {
logger.info("Waiting for Replicables " + Arrays.toString(replicableIdsAsStrings) +
" before firing up replication automatically...");
final List<Replicable<?, ?>> replicables = new ArrayList<>();
for (String replicableIdAsString : replicableIdsAsStrings) {
Replicable<?, ?> replicable = replicablesProvider.getReplicable(replicableIdAsString, /* wait */true);
logger.info("Obtained Replicable " + replicableIdAsString);
replicables.add(replicable);
}
}.start();
}
logger.info("Configuration requested automatic replication for replicables "+
Arrays.toString(replicableIdsAsStrings)+". Starting it up...");
String replicateFromExchangeName = System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_EXCHANGE_NAME);
if (replicateFromExchangeName == null) {
replicateFromExchangeName = masterExchangeName;
}
ReplicationMasterDescriptorImpl master = new ReplicationMasterDescriptorImpl(
System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_QUEUE_HOST),
replicateFromExchangeName,
Integer.valueOf(System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_QUEUE_PORT).trim()),
serverReplicationMasterService.getServerIdentifier().toString(),
System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_SERVLET_HOST),
Integer.valueOf(System.getProperty(PROPERTY_NAME_REPLICATE_MASTER_SERVLET_PORT).trim()), replicables);
try {
serverReplicationMasterService.startToReplicateFrom(master);
serverReplicationMasterService.setReplicationStarting(false);
logger.info("Automatic replication has been started.");
} catch (ClassNotFoundException | IOException | InterruptedException e) {
logger.log(Level.SEVERE, "Error with automatic replication from "+master, e);
}
}
}.start();
}
public void stop(BundleContext bundleContext) throws Exception {
@@ -45,6 +45,7 @@ import com.sap.sse.replication.ReplicablesProvider.ReplicableLifeCycleListener;
import com.sap.sse.replication.ReplicationMasterDescriptor;
import com.sap.sse.replication.ReplicationReceiver;
import com.sap.sse.replication.ReplicationService;
import com.sap.sse.replication.ReplicationStatus;
import com.sap.sse.util.HttpUrlConnectionHelper;
import net.jpountz.lz4.LZ4BlockInputStream;
@@ -840,4 +841,30 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
unsentOperationsSenderJob.scheduleForSending(operationWithResult, sender);
}
@Override
public synchronized ReplicationStatus getStatus() {
final ReplicationReceiver replicationReceiver = getReplicator();
final boolean isReplicationStarting = isReplicationStarting();
final boolean isReplica = isReplicationStarting || replicationReceiver != null;
final boolean suspended = replicationReceiver == null ? false : replicationReceiver.isSuspended();
final boolean stopped = replicationReceiver == null ? false : replicationReceiver.isBeingStopped();
long messageQueueLength;
if (replicationReceiver == null) {
messageQueueLength = 0;
} else {
try {
messageQueueLength = replicationReceiver.getMessageQueueSize();
} catch (IllegalAccessException e) {
logger.warning("Unable to access replication message queue size: "+e.getMessage()+". Reporting as -1");
messageQueueLength = -1;
}
}
final Map<String, Integer> operationQueueLengths = replicationReceiver == null ? new HashMap<>() : replicationReceiver.getOperationQueueSizes();
final Map<String, Boolean> isInitialLoadRunning = new HashMap<>();
for (final Replicable<?, ?> replicable : getAllReplicables()) {
isInitialLoadRunning.put(replicable.getId().toString(), replicable.isCurrentlyFillingFromInitialLoad());
}
return new ReplicationStatusImpl(isReplica, isReplicationStarting, suspended, stopped, messageQueueLength, isInitialLoadRunning, operationQueueLengths);
}
}
@@ -9,7 +9,6 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Date;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -29,8 +28,8 @@ import com.sap.sse.replication.OperationWithResult;
import com.sap.sse.replication.ReplicaDescriptor;
import com.sap.sse.replication.Replicable;
import com.sap.sse.replication.ReplicablesProvider;
import com.sap.sse.replication.ReplicationReceiver;
import com.sap.sse.replication.ReplicationService;
import com.sap.sse.replication.ReplicationStatus;
import com.sap.sse.util.impl.CountingOutputStream;
import net.jpountz.lz4.LZ4BlockInputStream;
@@ -182,41 +181,42 @@ public class ReplicationServlet extends AbstractHttpServlet {
private void reportStatus(HttpServletResponse resp) throws IllegalAccessException, IOException {
final JSONObject result = new JSONObject();
final JSONArray replicablesJSON = new JSONArray();
synchronized (getReplicationService()) {
final ReplicationReceiver replicationReceiver = getReplicationService().getReplicator();
final boolean isReplicationStarting = getReplicationService().isReplicationStarting();
final boolean isReplica = isReplicationStarting || replicationReceiver != null;
result.put("replica", isReplica);
result.put("replicationstarting", isReplicationStarting);
result.put("suspended", replicationReceiver == null ? false : replicationReceiver.isSuspended());
result.put("stopped", replicationReceiver == null ? false : replicationReceiver.isBeingStopped());
result.put("messagequeuelength", replicationReceiver == null ? 0 : replicationReceiver.getMessageQueueSize());
final JSONArray operationQueueLengths = new JSONArray();
result.put("operationqueuelengths", operationQueueLengths);
int totalOperationQueueLengths = 0;
if (replicationReceiver != null) {
for (final Entry<String, Integer> queueLength : replicationReceiver.getOperationQueueSizes().entrySet()) {
final JSONObject queueLengthJSON = new JSONObject();
queueLengthJSON.put("id", queueLength.getKey());
queueLengthJSON.put("length", queueLength.getValue());
totalOperationQueueLengths += queueLength.getValue();
operationQueueLengths.add(queueLengthJSON);
}
final ReplicationStatus status = getReplicationService().getStatus();
result.put("replica", status.isReplica());
result.put("replicationstarting", status.isReplicationStarting());
result.put("suspended", status.isSuspended());
result.put("stopped", status.isStopped());
result.put("messagequeuelength", status.getMessageQueueLength());
final JSONArray operationQueueLengths = new JSONArray();
result.put("operationqueuelengths", operationQueueLengths);
for (final String replicableIdAsString : status.getReplicableIdsAsStrings()) {
Integer queueLength = status.getOperationQueueLengthsByReplicableIdAsString(replicableIdAsString);
if (queueLength != null) {
final JSONObject queueLengthJSON = new JSONObject();
queueLengthJSON.put("id", replicableIdAsString);
queueLengthJSON.put("length", queueLength);
operationQueueLengths.add(queueLengthJSON);
}
result.put("totaloperationqueuelength", totalOperationQueueLengths);
boolean initialLoadRunning = false;
for (final Replicable<?, ?> replicable : getReplicationService().getAllReplicables()) {
}
result.put("totaloperationqueuelength", status.getTotalOperationQueueLength());
for (final String replicableIdAsString : status.getReplicableIdsAsStrings()) {
Boolean initialLoadRunning = status.isInitialLoadRunning(replicableIdAsString);
if (initialLoadRunning != null) {
final JSONObject replicableJSON = new JSONObject();
replicableJSON.put("id", replicable.getId());
replicableJSON.put("initialloadrunning", replicable.isCurrentlyFillingFromInitialLoad());
initialLoadRunning = initialLoadRunning || replicable.isCurrentlyFillingFromInitialLoad();
replicableJSON.put("id", replicableIdAsString);
replicableJSON.put("initialloadrunning", initialLoadRunning);
replicablesJSON.add(replicableJSON);
}
result.put("replicables", replicablesJSON);
result.put("available", !isReplica || (!isReplicationStarting && !initialLoadRunning));
}
result.put("replicables", replicablesJSON);
result.put("available", status.isAvailable());
resp.setContentType("application/json;charset=UTF-8");
resp.getWriter().print(result.toJSONString());
if (status.isAvailable()) {
resp.setStatus(HttpServletResponse.SC_OK);
} else {
resp.setStatus(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
}
}
/**
@@ -0,0 +1,133 @@
package com.sap.sse.replication.impl;
import java.util.Collections;
import java.util.Map;
import java.util.Map.Entry;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import com.sap.sse.replication.ReplicationStatus;
public class ReplicationStatusImpl implements ReplicationStatus {
private final boolean isReplica;
private final boolean isReplicationStarting;
private final boolean isSuspended;
private final boolean isStopped;
private final long messageQueueLength;
private final Map<String, Boolean> isInitialLoadRunning;
private final Map<String, Integer> operationQueueLengths;
public ReplicationStatusImpl(boolean isReplica, boolean isReplicationStarting, boolean isSuspended,
boolean isStopped, long messageQueueLength, Map<String, Boolean> isInitialLoadRunning,
Map<String, Integer> operationQueueLengths) {
super();
this.isReplica = isReplica;
this.isReplicationStarting = isReplicationStarting;
this.isSuspended = isSuspended;
this.isStopped = isStopped;
this.messageQueueLength = messageQueueLength;
this.isInitialLoadRunning = isInitialLoadRunning;
this.operationQueueLengths = operationQueueLengths;
}
@Override
public boolean isReplica() {
return isReplica;
}
@Override
public boolean isReplicationStarting() {
return isReplicationStarting;
}
@Override
public boolean isSuspended() {
return isSuspended;
}
@Override
public boolean isStopped() {
return isStopped;
}
@Override
public long getMessageQueueLength() {
return messageQueueLength;
}
@Override
public Iterable<String> getReplicableIdsAsStrings() {
return Collections.unmodifiableMap(isInitialLoadRunning).keySet();
}
@Override
public Integer getOperationQueueLengthsByReplicableIdAsString(String replicableIdAsString) {
return operationQueueLengths.get(replicableIdAsString);
}
@Override
public int getTotalOperationQueueLength() {
int result = 0;
for (Entry<String, Integer> e : operationQueueLengths.entrySet()) {
result += e.getValue();
}
return result;
}
@Override
public Boolean isInitialLoadRunning(String replicableIdAsString) {
return isInitialLoadRunning.get(replicableIdAsString);
}
@Override
public boolean isAvailable() {
return !isReplica() || (!isReplicationStarting() && !isInitialLoadRunning());
}
/**
* Tells whether the initial load is running for any of the replicables known
*/
private boolean isInitialLoadRunning() {
boolean result = false;
for (boolean initialLoadRunning : isInitialLoadRunning.values()) {
result |= initialLoadRunning;
}
return result;
}
@Override
public JSONObject toJSONObject() {
final JSONObject result = new JSONObject();
final JSONArray replicablesJSON = new JSONArray();
result.put("replica", this.isReplica());
result.put("replicationstarting", this.isReplicationStarting());
result.put("suspended", this.isSuspended());
result.put("stopped", this.isStopped());
result.put("messagequeuelength", this.getMessageQueueLength());
final JSONArray operationQueueLengths = new JSONArray();
result.put("operationqueuelengths", operationQueueLengths);
for (final String replicableIdAsString : this.getReplicableIdsAsStrings()) {
Integer queueLength = this.getOperationQueueLengthsByReplicableIdAsString(replicableIdAsString);
if (queueLength != null) {
final JSONObject queueLengthJSON = new JSONObject();
queueLengthJSON.put("id", replicableIdAsString);
queueLengthJSON.put("length", queueLength);
operationQueueLengths.add(queueLengthJSON);
}
}
result.put("totaloperationqueuelength", this.getTotalOperationQueueLength());
for (final String replicableIdAsString : this.getReplicableIdsAsStrings()) {
Boolean initialLoadRunning = this.isInitialLoadRunning(replicableIdAsString);
if (initialLoadRunning != null) {
final JSONObject replicableJSON = new JSONObject();
replicableJSON.put("id", replicableIdAsString);
replicableJSON.put("initialloadrunning", initialLoadRunning);
replicablesJSON.add(replicableJSON);
}
}
result.put("replicables", replicablesJSON);
result.put("available", this.isAvailable());
return result;
}
}
@@ -69,9 +69,9 @@ public class ReplicatingCache<K, V> implements Cache<K, V>, Named {
}
V result = cache.put(key, value);
final String myName = name;
securityService.replicate(s->
s.getCacheManager().getCache(myName).put(key, value));
if (store && value instanceof Session) {
securityService.replicate(s->
s.getCacheManager().getCache(myName).put(key, value));
securityService.storeSession(getName(), (Session) value);
}
return result;
+6 -7
View File
@@ -27,15 +27,14 @@ echo "JAVA_HOME: $JAVA_HOME"
echo "INSTANCE_ID: $INSTANCE_ID"
echo ""
RACING_EVENT_SERVICE_AVAILABLE=$(curl http://127.0.0.1:$SERVER_PORT/sailingserver/api/v1/status 2>/dev/null | jq .available)
REPLICATION_AVAILABLE=$(curl http://127.0.0.1:$SERVER_PORT/replication/replication?action=STATUS 2>/dev/null | jq .available)
INDEX_HTML_STATUS=$(curl -I http://127.0.0.1:8888/index.html 2>/dev/null | head -n 1 | cut -d ' ' -f 2)
STATUS=$(curl -i http://127.0.0.1:$SERVER_PORT/sailingserver/api/v1/status 2>/dev/null)
RESPONSE_STATUS_CODE=$(echo $STATUS | head -n 1 | cut -d ' ' -f 2)
echo "RACING_EVENT_SERVICE_AVAILABLE: $RACING_EVENT_SERVICE_AVAILABLE"
echo "REPLICATION_AVAILABLE: $REPLICATION_AVAILABLE"
echo "INDEX_HTML_STATUS: $INDEX_HTML_STATUS"
echo "AVAILABILITY_STATUS: $RESPONSE_STATUS_CODE"
echo "AVAILABILITY_DETAILS:"
echo "$STATUS" | grep "^\{" | jq .
if [[ $RACING_EVENT_SERVICE_AVAILABLE == "true" && $REPLICATION_AVAILABLE == "true" && $INDEX_HTML_STATUS =~ 2.. ]]; then
if [[ $RESPONSE_STATUS_CODE =~ 2.. ]]; then
echo "Server $SERVER_NAME is available on $SERVER_PORT"
exit 0
else