Implemented GET/GET Client and POST with replication for ML models

This commit is contained in:
Vladislav Chumak committed 2019-02-15 17:20:34 +01:00
1 parent 268b32f27d
commit d9a9114b98
16 files changed
+421 -38

No files matched your search

@@ -42,6 +42,7 @@ public enum Permission implements com.sap.sse.security.shared.Permission {
LEADERBOARD_GROUP,
TRACKED_RACE,
DATA_MINING,
WIND_ESTIMATION_MODELS
;
// TODO once we can use Java8 here, move this up into a "default" method on the Permission interface
@@ -43,7 +43,6 @@ public class RestServletContainer extends ServletContainer {
@Override
public void destroy() {
super.destroy();
if (polarDataServiceTracker != null) {
polarDataServiceTracker.close();
}
@@ -13,7 +13,7 @@ Require-Bundle: org.eclipse.osgi;bundle-version="3.10.1",
com.sap.sse;bundle-version="1.0.0",
com.sap.sse.common;bundle-version="1.0.0",
com.sap.sailing.domain.shared.android,
com.sap.sailing.domain.common;bundle-version="1.0.0",
com.sap.sailing.domain.common,
com.sap.sailing.server.gateway.serialization;bundle-version="1.0.0",
com.sap.sailing.server.gateway.serialization.shared.android;bundle-version="1.0.0",
com.sap.sailing.polars,
@@ -25,7 +25,23 @@ Require-Bundle: org.eclipse.osgi;bundle-version="3.10.1",
com.sap.sse.mongodb,
com.sap.sailing.domain,
com.sap.sse.replication.interfaces,
com.sap.sse.operationaltransformation
com.sap.sse.operationaltransformation,
com.sun.jersey,
javax.servlet;bundle-version="2.6.0",
org.apache.shiro.core;bundle-version="1.2.2",
org.apache.shiro.web;bundle-version="1.2.2",
org.apache.shiro.ehcache;bundle-version="1.2.3",
com.sun.xml.bind.jaxb-impl;bundle-version="2.3.0",
com.sap.sse.security,
com.sap.sse.security.common
Import-Package: javax.ws.rs;version="1.1.1",
javax.ws.rs.core;version="1.1.1",
javax.xml.bind,
org.apache.http,
org.apache.http.client;version="[4.2.6,4.3.0)",
org.apache.http.client.methods;version="[4.2.6,4.3.0)",
org.apache.http.impl.client;version="[4.2.6,4.3.0)"
Web-ContextPath: /windestimation
Bundle-RequiredExecutionEnvironment: JavaSE-1.8
Bundle-ActivationPolicy: lazy
Export-Package: com.sap.sailing.windestimation
@@ -0,0 +1,65 @@
<?xml version="1.0" encoding="utf-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://java.sun.com/xml/ns/javaee" xmlns:web="http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"
version="2.5">
<!-- Disable directory listing: -->
<servlet>
<servlet-name>default</servlet-name>
<servlet-class>org.eclipse.jetty.servlet.DefaultServlet</servlet-class>
<init-param>
<param-name>dirAllowed</param-name>
<param-value>false</param-value>
</init-param>
</servlet>
<!-- Apache Shiro -->
<context-param>
<param-name>shiroEnvironmentClass</param-name>
<param-value>org.apache.shiro.web.env.IniWebEnvironment</param-value>
</context-param>
<listener>
<listener-class>org.apache.shiro.web.env.EnvironmentLoaderListener</listener-class>
</listener>
<filter>
<filter-name>ShiroFilter</filter-name>
<filter-class>org.apache.shiro.web.servlet.ShiroFilter</filter-class>
</filter>
<!-- Make sure any request you want accessible to Shiro is filtered. "/*"
catches all requests. Usually this filter mapping is defined first (before
all others) to ensure that Shiro works in subsequent filters in the filter
chain: -->
<filter-mapping>
<filter-name>ShiroFilter</filter-name>
<url-pattern>/*</url-pattern>
<dispatcher>REQUEST</dispatcher>
<dispatcher>FORWARD</dispatcher>
<dispatcher>INCLUDE</dispatcher>
<dispatcher>ERROR</dispatcher>
</filter-mapping>
<servlet>
<servlet-name>Jersey REST API</servlet-name>
<servlet-class>com.sap.sailing.windestimation.jaxrs.RestServletContainer</servlet-class>
<init-param>
<param-name>javax.ws.rs.Application</param-name>
<param-value>com.sap.sailing.windestimation.jaxrs.api.RestApiApplication</param-value>
</init-param>
<init-param>
<param-name>com.sun.jersey.spi.container.ContainerResponseFilters</param-name>
<param-value>com.sap.sailing.windestimation.jaxrs.AccessControlAndEncodingResponseFilter</param-value>
</init-param>
<init-param>
<param-name>com.sun.jersey.config.property.packages</param-name>
<param-value>com.sap.sailing.windestimation.jaxrs</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
</servlet>
<servlet-mapping>
<servlet-name>Jersey REST API</servlet-name>
<url-pattern>/api/*</url-pattern>
</servlet-mapping>
</web-app>
@@ -4,6 +4,7 @@ import java.util.Dictionary;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.osgi.framework.BundleActivator;
@@ -11,6 +12,7 @@ import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceRegistration;
import com.sap.sailing.domain.windestimation.WindEstimationFactoryService;
import com.sap.sailing.windestimation.jaxrs.client.WindEstimationDataClient;
import com.sap.sailing.windestimation.model.exception.ModelLoadingException;
import com.sap.sailing.windestimation.model.exception.ModelPersistenceException;
import com.sap.sailing.windestimation.model.store.FileSystemModelStoreImpl;
@@ -54,8 +56,6 @@ public class Activator implements BundleActivator {
importWindEstimationModelsFromUrl(windEstimationModelDataSourceURL);
} else if (windEstimationModelDataSourceFolder != null) {
importWindEstimationModelsFromFolder(windEstimationModelDataSourceFolder);
} else {
importWindEstimationModelsFromMongoDb();
}
final ServiceRegistration<WindEstimationFactoryService> serviceRegistration = context
.registerService(WindEstimationFactoryService.class, service, null);
@@ -76,15 +76,6 @@ public class Activator implements BundleActivator {
}
}
private void importWindEstimationModelsFromMongoDb() {
MongoDbModelStoreImpl modelStore = new MongoDbModelStoreImpl(MongoDBService.INSTANCE.getDB());
try {
service.importAllModelsFromModelStore(modelStore);
} catch (ModelPersistenceException e) {
throw new ModelLoadingException("Could not import wind estimation models from MongoDB", e);
}
}
private void importWindEstimationModelsFromFolder(String windEstimationModelDataSourceFolder) {
FileSystemModelStoreImpl modelStore = new FileSystemModelStoreImpl(windEstimationModelDataSourceFolder);
try {
@@ -97,17 +88,14 @@ public class Activator implements BundleActivator {
private void importWindEstimationModelsFromUrl(String windEstimationModelDataSourceURL) {
logger.info("Importing wind estimation data from URL: " + windEstimationModelDataSourceURL);
// TODO implement import client for wind estimation models
// PolarDataClient polarDataClient = new PolarDataClient(windEstimationModelDataSourceURL,
// windEstimationFactoryService);
// try {
// polarDataClient.updatePolarDataRegressions();
// TODO ensure that polarDataService has been loaded and initialized
// } catch (Exception e) {
// logger.log(Level.SEVERE,
// "Exception while trying to import wind estimation model data from " +
// windEstimationModelDataSourceURL, e);
// }
WindEstimationDataClient polarDataClient = new WindEstimationDataClient(windEstimationModelDataSourceURL,
service);
try {
polarDataClient.updateWindEstimationModels();
} catch (Exception e) {
logger.log(Level.SEVERE, "Exception while trying to import wind estimation model data from "
+ windEstimationModelDataSourceURL, e);
}
logger.info("Import of wind estimation data finished successfully ");
}
@@ -0,0 +1,24 @@
package com.sap.sailing.windestimation.integration;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import com.sap.sailing.windestimation.model.store.ModelDomainType;
public class ExportedModels implements Serializable {
private static final long serialVersionUID = -2387862126027546463L;
private final Map<ModelDomainType, Map<String, byte[]>> serializedModelsPerDomainType = new HashMap<>();
public void addSerializedModelsForDomainType(ModelDomainType domainType,
Map<String, byte[]> serializedModelPerPersistenceKey) {
serializedModelsPerDomainType.put(domainType, serializedModelPerPersistenceKey);
}
public Map<String, byte[]> getSerializedModelsForDomainType(ModelDomainType domainType) {
return serializedModelsPerDomainType.get(domainType);
}
}
@@ -11,5 +11,5 @@ import com.sap.sse.replication.ReplicableWithObjectInputStream;
*
*/
public interface ReplicableWindEstimationFactoryService extends WindEstimationFactoryService,
ReplicableWithObjectInputStream<WindEstimationFactoryService, WindEstimationDataOperation<?>> {
ReplicableWithObjectInputStream<WindEstimationFactoryServiceImpl, WindEstimationModelsUpdateOperation> {
}
@@ -12,7 +12,6 @@ import com.sap.sailing.domain.common.WindSourceType;
import com.sap.sailing.domain.common.impl.WindSourceImpl;
import com.sap.sailing.domain.tracking.TrackedRace;
import com.sap.sailing.domain.windestimation.IncrementalWindEstimationTrack;
import com.sap.sailing.domain.windestimation.WindEstimationFactoryService;
import com.sap.sailing.windestimation.model.classifier.maneuver.ManeuverClassifiersCache;
import com.sap.sailing.windestimation.model.classifier.maneuver.ManeuverFeatures;
import com.sap.sailing.windestimation.model.exception.ModelPersistenceException;
@@ -20,9 +19,11 @@ import com.sap.sailing.windestimation.model.regressor.twdtransition.GaussianBase
import com.sap.sailing.windestimation.model.store.InMemoryModelStoreImpl;
import com.sap.sailing.windestimation.model.store.ModelDomainType;
import com.sap.sailing.windestimation.model.store.ModelStore;
import com.sap.sailing.windestimation.model.store.MongoDbModelStoreImpl;
import com.sap.sse.mongodb.MongoDBService;
public class WindEstimationFactoryServiceImpl
extends AbstractReplicableWithObjectInputStream<WindEstimationFactoryService, WindEstimationDataOperation<?>>
public class WindEstimationFactoryServiceImpl extends
AbstractReplicableWithObjectInputStream<WindEstimationFactoryServiceImpl, WindEstimationModelsUpdateOperation>
implements ReplicableWindEstimationFactoryService {
private static final boolean PRELOAD_ALL_MODELS = true;
@@ -45,13 +46,18 @@ public class WindEstimationFactoryServiceImpl
private boolean shutdown = false;
public WindEstimationFactoryServiceImpl() {
MODEL_STORE = new InMemoryModelStoreImpl();
MODEL_STORE = isMaster() ? new MongoDbModelStoreImpl(MongoDBService.INSTANCE.getDB())
: new InMemoryModelStoreImpl();
maneuverClassifiersCache = new ManeuverClassifiersCache(MODEL_STORE, PRELOAD_ALL_MODELS,
PRESERVE_LOADED_MODELS_MILLIS, MAX_MANEUVER_FEATURES);
gaussianBasedTwdTransitionDistributionCache = new GaussianBasedTwdTransitionDistributionCache(MODEL_STORE,
PRELOAD_ALL_MODELS, PRESERVE_LOADED_MODELS_MILLIS);
}
private boolean isMaster() {
return getMasterDescriptor() == null;
}
@Override
public IncrementalWindEstimationTrack createIncrementalWindEstimationTrack(TrackedRace trackedRace) {
IncrementalWindEstimationTrack windEstimation = new IncrementalMstHmmWindEstimationForTrackedRace(trackedRace,
@@ -66,13 +72,20 @@ public class WindEstimationFactoryServiceImpl
return modelsReady;
}
@SuppressWarnings("unchecked")
@Override
public void initiallyFillFromInternal(ObjectInputStream is)
throws IOException, ClassNotFoundException, InterruptedException {
ExportedModels exportedModels = (ExportedModels) is.readObject();
updateWindEstimationModels(exportedModels);
}
public void updateWindEstimationModels(ExportedModels exportedModels) throws ModelPersistenceException {
for (ModelDomainType domainType : modelDomainTypesRequiredByWindEstimation) {
Map<String, byte[]> exportedModels = (Map<String, byte[]>) is.readObject();
MODEL_STORE.importPersistedModels(exportedModels, domainType);
Map<String, byte[]> serializedModelsForDomainType = exportedModels
.getSerializedModelsForDomainType(domainType);
if (serializedModelsForDomainType != null) {
MODEL_STORE.importPersistedModels(serializedModelsForDomainType, domainType);
}
}
clearState();
}
@@ -106,10 +119,12 @@ public class WindEstimationFactoryServiceImpl
@Override
public void serializeForInitialReplicationInternal(ObjectOutputStream objectOutputStream) throws IOException {
ExportedModels exportedModels = new ExportedModels();
for (ModelDomainType domainType : modelDomainTypesRequiredByWindEstimation) {
Map<String, byte[]> exportedModels = MODEL_STORE.exportAllPersistedModels(domainType);
objectOutputStream.writeObject(exportedModels);
Map<String, byte[]> exportedModelsForDomainType = MODEL_STORE.exportAllPersistedModels(domainType);
exportedModels.addSerializedModelsForDomainType(domainType, exportedModelsForDomainType);
}
objectOutputStream.writeObject(exportedModels);
}
/**
@@ -1,6 +1,5 @@
package com.sap.sailing.windestimation.integration;
import com.sap.sailing.domain.windestimation.WindEstimationFactoryService;
import com.sap.sse.replication.OperationWithResult;
/**
@@ -9,8 +8,8 @@ import com.sap.sse.replication.OperationWithResult;
*
* @author Vladislav Chumak (D069712)
*
* @param <T>
*/
public interface WindEstimationDataOperation<T> extends OperationWithResult<WindEstimationFactoryService, T> {
public interface WindEstimationModelsUpdateOperation
extends OperationWithResult<WindEstimationFactoryServiceImpl, Void> {
}
@@ -0,0 +1,19 @@
package com.sap.sailing.windestimation.integration;
public class WindEstimationModelsUpdateOperationImpl implements WindEstimationModelsUpdateOperation {
private static final long serialVersionUID = 6407877837605630090L;
private final ExportedModels exportedModels;
public WindEstimationModelsUpdateOperationImpl(ExportedModels exportedModels) {
this.exportedModels = exportedModels;
}
@Override
public Void internalApplyTo(WindEstimationFactoryServiceImpl toState) throws Exception {
toState.updateWindEstimationModels(exportedModels);
return null;
}
}
@@ -0,0 +1,25 @@
package com.sap.sailing.windestimation.jaxrs;
import javax.servlet.ServletContext;
import javax.ws.rs.core.Context;
import org.osgi.util.tracker.ServiceTracker;
import com.sap.sailing.windestimation.integration.WindEstimationFactoryServiceImpl;
/**
*
* @author Vladislav Chumak (D069712)
*
*/
public abstract class AbstractWindEstimationDataResource {
@Context
ServletContext servletContext;
public WindEstimationFactoryServiceImpl getWindEstimationFactoryServiceImpl() {
@SuppressWarnings("unchecked")
ServiceTracker<WindEstimationFactoryServiceImpl, WindEstimationFactoryServiceImpl> tracker = (ServiceTracker<WindEstimationFactoryServiceImpl, WindEstimationFactoryServiceImpl>) servletContext
.getAttribute(RestServletContainer.WIND_ESTIMATION_FACTORY_SERVICE_TRACKER_NAME);
return tracker.getService();
}
}
@@ -0,0 +1,25 @@
package com.sap.sailing.windestimation.jaxrs;
import javax.ws.rs.core.MultivaluedMap;
import com.sun.jersey.spi.container.ContainerRequest;
import com.sun.jersey.spi.container.ContainerResponse;
import com.sun.jersey.spi.container.ContainerResponseFilter;
/**
*
* @author Vladislav Chumak (D069712)
*
*/
public class AccessControlAndEncodingResponseFilter implements ContainerResponseFilter {
@Override
public ContainerResponse filter(ContainerRequest request, ContainerResponse response) {
MultivaluedMap<String, Object> httpHeaders = response.getHttpHeaders();
httpHeaders.add("Access-Control-Allow-Origin", "*");
httpHeaders.add("Access-Control-Allow-Headers", "Authorization, Origin, X-Requested-With, Content-Type");
httpHeaders.add("Access-Control-Expose-Headers", "Location, Content-Disposition");
httpHeaders.add("Access-Control-Allow-Methods", "POST, PUT, GET, DELETE, HEAD, OPTIONS");
return response;
}
}
@@ -0,0 +1,59 @@
package com.sap.sailing.windestimation.jaxrs;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.ws.rs.core.Application;
import org.osgi.framework.BundleContext;
import org.osgi.util.tracker.ServiceTracker;
import com.sap.sailing.domain.windestimation.WindEstimationFactoryService;
import com.sun.jersey.spi.container.servlet.ServletContainer;
/**
*
* @author Vladislav Chumak (D069712)
*
*/
public class RestServletContainer extends ServletContainer {
private static final long serialVersionUID = 3129261471808425410L;
private static final String OSGI_RFC66_WEBBUNDLE_BUNDLECONTEXT_NAME = "osgi-bundlecontext";
public static final String WIND_ESTIMATION_FACTORY_SERVICE_TRACKER_NAME = "windEstimationFactoryServiceTracker";
private ServiceTracker<WindEstimationFactoryService, WindEstimationFactoryService> windEstimationFactoryServiceTracker;
public RestServletContainer() {
super();
}
public RestServletContainer(Application app) {
super(app);
}
public RestServletContainer(Class<? extends Application> appClass) {
super(appClass);
}
@Override
public void init(ServletConfig config) throws ServletException {
super.init(config);
BundleContext context = (BundleContext) config.getServletContext()
.getAttribute(OSGI_RFC66_WEBBUNDLE_BUNDLECONTEXT_NAME);
windEstimationFactoryServiceTracker = new ServiceTracker<WindEstimationFactoryService, WindEstimationFactoryService>(
context, WindEstimationFactoryService.class.getName(), null);
windEstimationFactoryServiceTracker.open();
config.getServletContext().setAttribute(WIND_ESTIMATION_FACTORY_SERVICE_TRACKER_NAME,
windEstimationFactoryServiceTracker);
}
@Override
public void destroy() {
super.destroy();
if (windEstimationFactoryServiceTracker != null) {
windEstimationFactoryServiceTracker.close();
}
}
}
@@ -0,0 +1,25 @@
package com.sap.sailing.windestimation.jaxrs.api;
import java.util.HashSet;
import java.util.Set;
import javax.ws.rs.core.Application;
import com.sap.sse.security.jaxrs.ShiroAuthorizationExceptionTo401ResponseMapper;
/**
*
* @author Vladislav Chumak (D069712)
*
*/
public class RestApiApplication extends Application {
public Set<Class<?>> getClasses() {
HashSet<Class<?>> classes = new HashSet<>();
// RESTlets
classes.add(WindEstimationDataResource.class);
// Exception Mapper
classes.add(ShiroAuthorizationExceptionTo401ResponseMapper.class);
return classes;
}
}
@@ -0,0 +1,53 @@
package com.sap.sailing.windestimation.jaxrs.api;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Response;
import org.apache.shiro.SecurityUtils;
import com.sap.sailing.domain.common.security.Permission;
import com.sap.sailing.domain.common.security.Permission.Mode;
import com.sap.sailing.windestimation.integration.ExportedModels;
import com.sap.sailing.windestimation.integration.WindEstimationModelsUpdateOperation;
import com.sap.sailing.windestimation.integration.WindEstimationModelsUpdateOperationImpl;
import com.sap.sailing.windestimation.jaxrs.AbstractWindEstimationDataResource;
/**
*
* @author Vladislav Chumak (D069712)
*
*/
@Path("/windestimation_data")
public class WindEstimationDataResource extends AbstractWindEstimationDataResource {
@GET
@Produces("application/octet-stream;charset=UTF-8")
public Response getInternalModelData() throws IOException {
final ByteArrayOutputStream bos = new ByteArrayOutputStream();
getWindEstimationFactoryServiceImpl().serializeForInitialReplication(bos);
bos.close();
return Response.ok(new ByteArrayInputStream(bos.toByteArray()))
.header("Content-Type", "application/octet-stream").build();
}
@POST
@Produces("text/plain")
public Response postInternalModelData(InputStream inputStream) throws Exception {
SecurityUtils.getSubject().checkPermission(Permission.WIND_ESTIMATION_MODELS.getStringPermission(Mode.UPDATE));
ObjectInputStream ois = getWindEstimationFactoryServiceImpl()
.createObjectInputStreamResolvingAgainstCache(inputStream);
ExportedModels exportedModels = (ExportedModels) ois.readObject();
WindEstimationModelsUpdateOperation windEstimationModelsUpdateOperation = new WindEstimationModelsUpdateOperationImpl(
exportedModels);
getWindEstimationFactoryServiceImpl().apply(windEstimationModelsUpdateOperation);
return Response.ok("Wind estimation models accepted").build();
}
}
@@ -0,0 +1,70 @@
package com.sap.sailing.windestimation.jaxrs.client;
import java.io.IOException;
import java.io.InputStream;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.SystemDefaultHttpClient;
import org.json.simple.parser.ParseException;
import com.sap.sailing.windestimation.integration.ReplicableWindEstimationFactoryService;
import com.sap.sailing.windestimation.model.store.ModelStore;
/**
* This class is used to replicate wind estimation models from the remote server using Apache {@link HttpClient}.
*
* @author Vladislav Chumak (D069712)
*
*/
public class WindEstimationDataClient {
private static final Logger logger = Logger.getLogger(WindEstimationDataClient.class.getName());
private static final String RESOURCE = "windestimation/api/windestimation_data";
private final ReplicableWindEstimationFactoryService windEstimationFactoryService;
private final String windEstimationDataSourceURL;
public WindEstimationDataClient(String windEstimationDataSourceURL,
ReplicableWindEstimationFactoryService windEstimationFactoryService) {
this.windEstimationFactoryService = windEstimationFactoryService;
this.windEstimationDataSourceURL = windEstimationDataSourceURL;
}
/**
* This method is used to update {@link ModelStore} of wind estimation with data received from remote server.
*/
public void updateWindEstimationModels()
throws IOException, ParseException, InterruptedException, ClassNotFoundException {
try {
logger.log(Level.INFO,
"Loading model data of wind estimation from remote server " + windEstimationDataSourceURL);
final InputStream inputStream = getContentFromResponse();
windEstimationFactoryService.clearReplicaState();
windEstimationFactoryService.initiallyFillFrom(inputStream);
logger.log(Level.INFO, "Loading model data of wind estimation from remote server "
+ windEstimationDataSourceURL + " succeeded");
} catch (Exception e) {
// Catching ClientProtocolException to indicate problems with HTTP protocol
logger.log(Level.WARNING, "Failed to load model data of wind estimation from remote server "
+ windEstimationDataSourceURL + ", " + e.getMessage());
windEstimationFactoryService.clearReplicaState();
}
}
private String getAPIString() {
return windEstimationDataSourceURL + (windEstimationDataSourceURL.endsWith("/") ? "" : "/") + RESOURCE;
}
protected InputStream getContentFromResponse() throws IOException, ParseException {
HttpClient client = new SystemDefaultHttpClient();
HttpGet getProcessor = new HttpGet(getAPIString());
HttpResponse processorResponse = client.execute(getProcessor);
return processorResponse.getEntity().getContent();
}
}