eclipse-sailing-analytics/sailing-analytics
1
0
Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-09-22 21:55:39 +00:00
Code Activity

bug5440: improved implementation of FixIngestionLambda and FixCombinationLambda to use a central class S3FixStorageStructure organizing the S3

Browse Source
This commit is contained in:
Kevin Wiesner
2022-09-30 12:41:39 +02:00
parent 32143eb633
commit cba9c545b4
4 changed files with 494 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
java/com.sap.sailing.ingestion/src/main/java/com/sap/sailing/ingestion/FixCombinationLambda.java
+154 -131
View File
@@ -3,17 +3,13 @@ package com.sap.sailing.ingestion;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.sql.Date;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.TreeSet;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.stream.Collectors;
@@ -25,15 +21,14 @@ import org.json.simple.parser.ParseException;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
import com.sap.sailing.domain.common.DeviceIdentifier;
import com.sap.sailing.domain.common.tracking.GPSFixMoving;
import com.sap.sailing.domain.tracking.impl.TimedComparator;
import com.sap.sailing.server.gateway.deserialization.impl.FlatSmartphoneUuidAndGPSFixMovingJsonDeserializer;
import com.sap.sailing.server.gateway.deserialization.impl.GPSFixMovingJsonDeserializer;
import com.sap.sailing.server.gateway.deserialization.impl.Helpers;
import com.sap.sailing.server.gateway.serialization.impl.GPSFixMovingJsonSerializer;
import com.sap.sse.common.Util.Pair;
import com.sap.sse.common.TimeRange;
import com.sap.sse.shared.json.JsonDeserializationException;
import com.sap.sse.shared.json.JsonDeserializer;
import software.amazon.awssdk.core.ResponseInputStream;
import software.amazon.awssdk.core.sync.RequestBody;
@@ -51,21 +46,31 @@ import software.amazon.awssdk.services.s3.model.S3Object;
import software.amazon.awssdk.services.s3.paginators.ListObjectsV2Iterable;
import software.amazon.awssdk.utils.IoUtils;
/**
* This λ is used to combine single fixes (currently only fixes from the type {@link GPSFixMoving}) on the S3 created by
* the {@link FixIngestionLambda} to collection items which bundle all single fixes for a certain device and a certain
* {@link TimeRange} obtained by the S3 structure class {@link S3FixStorageStructure}.
*
* @author Kevin Wiesner
*
*/
public class FixCombinationLambda implements RequestStreamHandler {
private static final Logger logger = Logger.getLogger(FixCombinationLambda.class.getName());
private final S3Client s3Client = S3Client.builder().region(Configuration.S3_REGION).build();
private final JsonDeserializer<Pair<UUID, List<GPSFixMoving>>> deserializer = new FlatSmartphoneUuidAndGPSFixMovingJsonDeserializer();
private final GPSFixMovingJsonDeserializer deserializerNoUUID = new GPSFixMovingJsonDeserializer();
private final GPSFixMovingJsonDeserializer deserializer = new GPSFixMovingJsonDeserializer();
private final S3FixStorageStructure s3FixStorageStructure = new S3FixStorageStructure();
private static final Logger logger = Logger.getLogger(FixCombinationLambda.class.getName());
/**
* Entry point for the λ on AWS. Defines the operation needed to combine single fixes to collections.
*/
@Override
public void handleRequest(final InputStream input, final OutputStream output, final Context context) {
try {
logger.info("FixCombination Lambda is starting");
Map<String, List<S3Object>> allNewFixMetadataFromDevices = this.getMetadataOfNewFixes();
ArrayList<ObjectIdentifier> keysToDelete = new ArrayList<ObjectIdentifier>();
Map<String, TreeSet<GPSFixMoving>> allFixDataFromDevices = this
.getDataOfNewFixes(allNewFixMetadataFromDevices, keysToDelete);
this.divideFixesAndCreateCollections(allFixDataFromDevices);
final Map<DeviceIdentifier, List<S3Object>> allSingleFixMetadataFromDevices = this.getMetadataOfNewFixes();
final List<ObjectIdentifier> keysToDelete = Collections.synchronizedList(new ArrayList<ObjectIdentifier>());
final Map<DeviceIdentifier, TreeSet<GPSFixMoving>> allSingleFixDataFromDevices = this.getDataOfNewFixes(allSingleFixMetadataFromDevices, keysToDelete);
this.divideFixesAndCreateCollections(allSingleFixDataFromDevices);
this.deleteFixesUsedForCollections(keysToDelete);
} catch (S3Exception e) {
logger.log(Level.SEVERE, e.awsErrorDetails().errorMessage());
@@ -83,148 +88,166 @@ public class FixCombinationLambda implements RequestStreamHandler {
}
}
private Map<String, List<S3Object>> getMetadataOfNewFixes() {
final ListObjectsV2Request listNewFixesRequest = ListObjectsV2Request.builder()
.bucket(Configuration.S3_BUCKET_NAME).prefix("testCombination/").delimiter("/collection/").build();
final ListObjectsV2Iterable listNewFixes = s3Client.listObjectsV2Paginator(listNewFixesRequest);
Map<String, List<S3Object>> allNewFixMetadataFromDevices = listNewFixes.stream()
.flatMap(listNewFixesPage -> listNewFixesPage.contents().stream())
.filter(newFix -> newFix.key().contains(".json"))
.collect(Collectors.groupingBy(newFix -> newFix.key().substring(0, newFix.key().lastIndexOf("/"))));
return allNewFixMetadataFromDevices;
/**
* Obtains the metadata for every single fix located on the S3.
*
* @return allSingleFixMetadataFromDevices
*/
private Map<DeviceIdentifier, List<S3Object>> getMetadataOfNewFixes() {
final ListObjectsV2Request listSingleFixesRequest = ListObjectsV2Request.builder()
.bucket(Configuration.S3_BUCKET_NAME)
.prefix(s3FixStorageStructure.getSingleFixPrefix())
.build();
final ListObjectsV2Iterable listOfSingleFixes = s3Client.listObjectsV2Paginator(listSingleFixesRequest);
final Map<DeviceIdentifier, List<S3Object>> allSingleFixMetadataFromDevices = listOfSingleFixes.stream()
.flatMap(listOfSingleFixesPage -> listOfSingleFixesPage.contents().stream())
.filter(singleFix -> singleFix.key().contains(".json"))
.collect(Collectors.groupingBy(singleFix -> s3FixStorageStructure.getIdentifierFromKey(singleFix.key())));
return allSingleFixMetadataFromDevices;
}
private Map<String, TreeSet<GPSFixMoving>> getDataOfNewFixes(
Map<String, List<S3Object>> allNewFixMetadataFromDevices, ArrayList<ObjectIdentifier> keysToDelete) {
Map<String, TreeSet<GPSFixMoving>> allFixDataFromDevices = new HashMap<String, TreeSet<GPSFixMoving>>();
allNewFixMetadataFromDevices.entrySet().parallelStream().forEach(newFixMetadataFromDevice -> {
logger.info("Fetching fixes from: " + newFixMetadataFromDevice.getKey() + " (Amount: "
+ Integer.toString(newFixMetadataFromDevice.getValue().size()) + ")");
allFixDataFromDevices.put(newFixMetadataFromDevice.getKey(), new TreeSet<>(new TimedComparator()));
newFixMetadataFromDevice.getValue().parallelStream().forEach(newFix -> {
GetObjectRequest getDataForSingleFixRequest = GetObjectRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME).key(newFix.key()).build();
/**
* Obtains the data from every single fix located on the S3 based on the given collection of metadata of single
* fixes. It returns a {@link Map<DeviceIdentifier, TreeSet<GPSFixMoving>>} which groups the single fixes to a
* certain {@link DeviceIdentifier} and organizes the fixes in a sorted {@link TreeSet<GPSFixMoving>} based on the
* timestamp of a single fix.
*
* @param allSingleFixMetadataFromDevices
* @param keysToDelete
* @return allSingleFixDataFromDevices
*/
private Map<DeviceIdentifier, TreeSet<GPSFixMoving>> getDataOfNewFixes(
final Map<DeviceIdentifier, List<S3Object>> allSingleFixMetadataFromDevices,
final List<ObjectIdentifier> keysToDelete) {
final Map<DeviceIdentifier, TreeSet<GPSFixMoving>> allSingleFixDataFromDevices = new HashMap<DeviceIdentifier, TreeSet<GPSFixMoving>>();
allSingleFixMetadataFromDevices.entrySet().parallelStream().forEach(singleFixMetadataFromDevice -> {
logger.info("Fetching fixes from: " + singleFixMetadataFromDevice.getKey()
+ " (Amount: " + Integer.toString(singleFixMetadataFromDevice.getValue().size()) + ")");
synchronized (allSingleFixDataFromDevices) {
allSingleFixDataFromDevices.put(singleFixMetadataFromDevice.getKey(), new TreeSet<>(new TimedComparator()));
}
singleFixMetadataFromDevice.getValue().parallelStream().forEach(singleFixS3Object -> {
try {
final Object newFixObject = JSONValue.parseWithException(
FixCombinationLambda.loadJSONFromGetRequest(s3Client, getDataForSingleFixRequest));
final JSONObject newFixJson = Helpers.toJSONObjectSafe(newFixObject);
final Pair<UUID, List<GPSFixMoving>> data = deserializer.deserialize(newFixJson);
final List<GPSFixMoving> fixes = data.getB();
for (GPSFixMoving fix : fixes) {
if (this.moreThanTenMinutesAgo(fix)) {
allFixDataFromDevices.get(newFixMetadataFromDevice.getKey()).add(fix);
final ObjectIdentifier fixToDelete = ObjectIdentifier.builder().key(newFix.key()).build();
keysToDelete.add(fixToDelete);
final Object singleFixObject = loadObjectFromKey(singleFixS3Object.key());
final JSONObject singleFixJson = Helpers.toJSONObjectSafe(singleFixObject);
final GPSFixMoving singleFix = deserializer.deserialize(singleFixJson);
if (!s3FixStorageStructure.isFixInCurrentCollectionFrame(singleFix)) {
synchronized (allSingleFixDataFromDevices) {
allSingleFixDataFromDevices.get(singleFixMetadataFromDevice.getKey()).add(singleFix);
}
final ObjectIdentifier fixToDelete = ObjectIdentifier.builder().key(singleFixS3Object.key()).build();
keysToDelete.add(fixToDelete);
}
} catch (ParseException | JsonDeserializationException e) {
logger.log(Level.SEVERE,
"Exception for key " + newFix.key() + " with JSON operations: " + e.getMessage());
logger.log(Level.SEVERE, "Exception for key " + singleFixS3Object.key() + " with JSON operations: " + e.getMessage());
} catch (IOException e) {
logger.log(Level.SEVERE,
"Exception for key " + newFix.key() + " while receiving object from S3: " + e.getMessage());
logger.log(Level.SEVERE, "Exception for key " + singleFixS3Object.key() + " while receiving object from S3: " + e.getMessage());
}
});
});
logger.info("Finished list of objects in S3");
return allFixDataFromDevices;
return allSingleFixDataFromDevices;
}
private void divideFixesAndCreateCollections(Map<String, TreeSet<GPSFixMoving>> allFixDataFromDevices) {
allFixDataFromDevices.entrySet().parallelStream().forEach(fixDataFromDevice -> {
fixDataFromDevice.getValue().stream().filter(this::moreThanTenMinutesAgo)
.collect(Collectors.groupingBy(this::roundToTenMinutes,
Collectors.toCollection(() -> new TreeSet<>(new TimedComparator()))))
.forEach((timestamp, newFixesTimeSet) -> {
String logAboutCollectionProcess = "Device Id: "
+ String.join("-", fixDataFromDevice.getKey().split("/")) + "\tTimestamp: "
+ Long.toString(timestamp) + "\tAmount of new Fixes: "
+ Integer.toString(newFixesTimeSet.size());
logAboutCollectionProcess += "\tFixes: " + newFixesTimeSet.toString();
TreeSet<GPSFixMoving> fixSetToStore = null;
final String keyForCollection = this.generateKeyForCollection(fixDataFromDevice.getKey(),
timestamp);
final GetObjectRequest existingFixCollectionRequest = GetObjectRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME).key(keyForCollection).build();
try {
final Object fixObject = JSONValue.parseWithException(FixCombinationLambda
.loadJSONFromGetRequest(s3Client, existingFixCollectionRequest));
final JSONArray fixJsonArray = Helpers.toJSONArraySafe(fixObject);
TreeSet<GPSFixMoving> existingFixes = new TreeSet<>(new TimedComparator());
Iterator<Object> fixJsonArrayIterator = fixJsonArray.iterator();
while (fixJsonArrayIterator.hasNext()) {
final JSONObject fixJsonObject = Helpers.toJSONObjectSafe(fixJsonArrayIterator.next());
final GPSFixMoving data = deserializerNoUUID.deserialize(fixJsonObject);
existingFixes.add(data);
}
if (existingFixes.size() > newFixesTimeSet.size()) {
existingFixes.addAll(newFixesTimeSet);
fixSetToStore = existingFixes;
} else {
newFixesTimeSet.addAll(existingFixes);
}
logAboutCollectionProcess += "\tAdded data to existing collection with key "
+ keyForCollection;
} catch (NoSuchKeyException e) {
logAboutCollectionProcess += "\tCollection with key " + keyForCollection
+ " does not exist, creating a new one...";
} catch (ParseException | JsonDeserializationException e) {
logAboutCollectionProcess += "\tDeserializing existing collection for key "
+ keyForCollection + " failed: " + e.getMessage();
} catch (IOException e) {
logAboutCollectionProcess += "\tFetching existing collection for key " + keyForCollection
+ " failed: " + e.getMessage();
}
final PutObjectRequest saveFixCollectionRequest = PutObjectRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME).key(keyForCollection).build();
if (fixSetToStore == null)
fixSetToStore = newFixesTimeSet;
JSONArray fixSetToStoreAsJson = new JSONArray();
fixSetToStore.stream().map((object) -> new GPSFixMovingJsonSerializer().serialize(object))
.forEach(fixSetToStoreAsJson::add);
s3Client.putObject(saveFixCollectionRequest,
RequestBody.fromString(fixSetToStoreAsJson.toJSONString()));
logger.info(logAboutCollectionProcess);
});
/**
* Uses a given map of all single fix data grouped by their device to first group every set for every device into
* time ranges for the resulting collections. If a collection already exists, it retrieves the single fixes them and
* sorts the new fixes for the collection into them. If the resulting collection does not yet exist, it creates a
* new sorted collection for the considered fixes.
*
* @param allSingleFixDataFromDevices
*/
private void divideFixesAndCreateCollections(Map<DeviceIdentifier, TreeSet<GPSFixMoving>> allSingleFixDataFromDevices) {
allSingleFixDataFromDevices.entrySet().parallelStream().forEach(singleFixDataFromDevice -> {
Map<TimeRange, TreeSet<GPSFixMoving>> timeRangeDividedFixDataFromDevice = singleFixDataFromDevice.getValue().stream()
.collect(Collectors.groupingBy(s3FixStorageStructure::assignFixToCollectionTimeUnit,
Collectors.toCollection(() -> new TreeSet<>(new TimedComparator()))));
for(Map.Entry<TimeRange, TreeSet<GPSFixMoving>> singleTimeRangeDataOfSingleDevice: timeRangeDividedFixDataFromDevice.entrySet()) {
final TimeRange timeRangeOfCollection = singleTimeRangeDataOfSingleDevice.getKey();
final TreeSet<GPSFixMoving> singleFixesMappedToTimeRange = singleTimeRangeDataOfSingleDevice.getValue();
String logAboutCollectionProcess = "Device Id: "
+ singleFixDataFromDevice.getKey().getStringRepresentation() + "\tTimestamp: "
+ timeRangeOfCollection.toString() + "\tAmount of new Fixes: "
+ Integer.toString(singleFixesMappedToTimeRange.size())
+ "\tFixes: " + singleFixesMappedToTimeRange.toString();
TreeSet<GPSFixMoving> fixSetToStoreForTimeRangeAndDevice = null;
final String keyForCollection = s3FixStorageStructure.generateKeyForCollection(singleFixDataFromDevice.getKey(), timeRangeOfCollection);
try {
final Object existingFixesObject = loadObjectFromKey(keyForCollection);
final JSONArray existingFixedJsonArray = Helpers.toJSONArraySafe(existingFixesObject);
TreeSet<GPSFixMoving> existingFixes = new TreeSet<>(new TimedComparator());
Iterator<Object> existingFixesJsonArrayIterator = existingFixedJsonArray.iterator();
while (existingFixesJsonArrayIterator.hasNext()) {
final JSONObject fixJsonObject = Helpers.toJSONObjectSafe(existingFixesJsonArrayIterator.next());
final GPSFixMoving existingSingleFixData = deserializer.deserialize(fixJsonObject);
existingFixes.add(existingSingleFixData);
}
if (existingFixes.size() > singleFixesMappedToTimeRange.size()) {
existingFixes.addAll(singleFixesMappedToTimeRange);
fixSetToStoreForTimeRangeAndDevice = existingFixes;
} else {
singleFixesMappedToTimeRange.addAll(existingFixes);
}
logAboutCollectionProcess += "\tAdded data to existing collection with key " + keyForCollection;
} catch (NoSuchKeyException e) {
logAboutCollectionProcess += "\tCollection with key " + keyForCollection + " does not exist, creating a new one...";
} catch (ParseException | JsonDeserializationException e) {
logAboutCollectionProcess += "\tDeserializing existing collection for key " + keyForCollection + " failed: " + e.getMessage();
} catch (IOException e) {
logAboutCollectionProcess += "\tFetching existing collection for key " + keyForCollection + " failed: " + e.getMessage();
}
final PutObjectRequest saveFixCollectionRequest = PutObjectRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME)
.key(keyForCollection)
.build();
if (fixSetToStoreForTimeRangeAndDevice == null)
fixSetToStoreForTimeRangeAndDevice = singleFixesMappedToTimeRange;
final JSONArray fixSetToStoreForTimeRangeAndDeviceAsJSONArray = new JSONArray();
fixSetToStoreForTimeRangeAndDevice.stream().map((fixObject) -> new GPSFixMovingJsonSerializer().serialize(fixObject)).forEach(fixSetToStoreForTimeRangeAndDeviceAsJSONArray::add);
s3Client.putObject(saveFixCollectionRequest, RequestBody.fromString(fixSetToStoreForTimeRangeAndDeviceAsJSONArray.toJSONString()));
logger.info(logAboutCollectionProcess);
}
});
logger.info("Finished combination of fixes into collection files");
}
private void deleteFixesUsedForCollections(ArrayList<ObjectIdentifier> keysToDelete) {
/**
* Deletes all the specified keys (given as a list of {@link ObjectIdentifier}) from the S3.
*
* @param keysToDelete
*/
private void deleteFixesUsedForCollections(final List<ObjectIdentifier> keysToDelete) {
logger.info("Keys to delete: " + keysToDelete.toString());
if (keysToDelete.size() != 0) {
final Delete deleteFixes = Delete.builder().objects(keysToDelete).build();
final DeleteObjectsRequest deleteFixesRequest = DeleteObjectsRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME).delete(deleteFixes).build();
.bucket(Configuration.S3_BUCKET_NAME)
.delete(deleteFixes)
.build();
s3Client.deleteObjects(deleteFixesRequest);
logger.info("Deleted single fix files used for collection");
}
}
private long roundToTenMinutes(GPSFixMoving fixToRound) {
final Timestamp t = new Timestamp(fixToRound.getTimePoint().asMillis());
final long fixTime = t.getTime();
t.setTime(fixTime - (fixTime % 600_000));
return t.getTime();
}
private Boolean moreThanTenMinutesAgo(GPSFixMoving fixToCheck) {
final long t = fixToCheck.getTimePoint().asMillis();
return t + 600_000 < System.currentTimeMillis();
}
private String generateKeyForCollection(String deviceKey, long timestampOfFix) {
final SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd/HH:mm");
sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
final Date d = new Date(timestampOfFix);
return deviceKey + "/collection/" + sdf.format(d) + "-UTC.json";
}
protected static String loadJSONFromGetRequest(S3Client s3Client, GetObjectRequest getObjectRequest)
throws IOException {
/**
* Helper function to retrieve data of a given S3 key and parse it to a Object which can then be used for JSON
* deserialization.
*
* @param requestKey
* @return fixObject
* @throws IOException
* @throws ParseException
*/
private Object loadObjectFromKey(final String requestKey)
throws IOException, ParseException {
final GetObjectRequest getObjectRequest = GetObjectRequest.builder()
.bucket(Configuration.S3_BUCKET_NAME)
.key(requestKey)
.build();
final ResponseInputStream<GetObjectResponse> fixData = s3Client.getObject(getObjectRequest);
final String strFixData = IoUtils.toUtf8String(fixData);
fixData.close();
return strFixData;
final Object fixObject = JSONValue.parseWithException(strFixData);
return fixObject;
}
}
java/com.sap.sailing.ingestion/src/main/java/com/sap/sailing/ingestion/FixIngestionLambda.java
+59 -32
View File
@@ -7,22 +7,37 @@ import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.json.simple.JSONObject;
import org.json.simple.JSONValue;
import org.json.simple.parser.ParseException;
import org.redisson.api.RMap;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestStreamHandler;
import com.google.gson.Gson;
import com.sap.sailing.domain.common.DeviceIdentifier;
import com.sap.sailing.domain.common.tracking.GPSFixMoving;
import com.sap.sailing.domain.racelogtracking.impl.SmartphoneUUIDIdentifierImpl;
import com.sap.sailing.ingestion.dto.AWSRequestWrapper;
import com.sap.sailing.ingestion.dto.AWSResponseWrapper;
import com.sap.sailing.ingestion.dto.EndpointDTO;
import com.sap.sailing.ingestion.dto.FixHeaderDTO;
import com.sap.sailing.ingestion.dto.GpsFixPayloadDTO;
import com.sap.sailing.server.gateway.deserialization.impl.FlatSmartphoneUuidAndGPSFixMovingJsonDeserializer;
import com.sap.sailing.server.gateway.deserialization.impl.Helpers;
import com.sap.sailing.server.gateway.serialization.impl.GPSFixMovingJsonSerializer;
import com.sap.sse.common.Util.Pair;
import com.sap.sse.shared.json.JsonDeserializationException;
import com.sap.sse.shared.json.JsonDeserializer;
import com.sap.sse.shared.json.JsonSerializer;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
@@ -34,30 +49,44 @@ import software.amazon.awssdk.utils.IoUtils;
* {@link AWSRequestWrapper}. In most cases clients will want to submit GPS fixes thus adhering to the
* {@link GpsFixPayloadDTO} structure. This structure will be recognized by most sailing servers.
*/
//TODO remove Gson dependency
public class FixIngestionLambda implements RequestStreamHandler {
final S3Client s3Client = S3Client.builder().region(Configuration.S3_REGION).build();
final RMap<String, List<EndpointDTO>> cacheMap = RedisUtils.getCacheMap();
private final JsonDeserializer<Pair<UUID, List<GPSFixMoving>>> deserializer =
new FlatSmartphoneUuidAndGPSFixMovingJsonDeserializer();
private final JsonSerializer<GPSFixMoving> serializer = new GPSFixMovingJsonSerializer();
private final S3FixStorageStructure s3FixStorageStructure = new S3FixStorageStructure();
private static final Logger logger = Logger.getLogger(FixIngestionLambda.class.getName());
@Override
public void handleRequest(final InputStream input, final OutputStream output, final Context context) {
try {
logger.info("Starting Lambda");
final byte[] streamAsBytes = IoUtils.toByteArray(input);
context.getLogger().log(new String(streamAsBytes));
logger.info("Input: " + new String(streamAsBytes));
final AWSRequestWrapper dtoWrapped = new Gson().fromJson(new String(streamAsBytes), AWSRequestWrapper.class);
final FixHeaderDTO dto = dtoWrapped.getBodyAsType(FixHeaderDTO.class);
Object requestBody = JSONValue.parseWithException(dtoWrapped.getBody());
JSONObject requestObject = Helpers.toJSONObjectSafe(requestBody);
Pair<UUID, List<GPSFixMoving>> data = deserializer.deserialize(requestObject);
DeviceIdentifier deviceIdentifier = new SmartphoneUUIDIdentifierImpl(data.getA());
List<GPSFixMoving> newFixes = data.getB();
final byte[] bodyAsBytes = dtoWrapped.getBody().getBytes();
final ForkJoinPool dispatchToSubscribersTask = ForkJoinPool.commonPool();
dispatchToSubscribersTask.submit(() -> {
try {
storeFixFileToS3(dto.getDeviceUuid(), bodyAsBytes, context.getLogger());
storeFixFileToS3(deviceIdentifier, newFixes);
} catch (IOException e) {
context.getLogger().log("Exception trying to store fixes to S3: "+e.getMessage());
logger.log(Level.SEVERE, "Exception trying to store fixes to S3: "+e.getMessage());
}
});
final RMap<String, List<EndpointDTO>> cacheMap = RedisUtils.getCacheMap();
final List<EndpointDTO> listOfEndpointsToTrigger = cacheMap.get(dto.getDeviceUuid());
final List<EndpointDTO> listOfEndpointsToTrigger = cacheMap.get(deviceIdentifier.getStringRepresentation());
if (listOfEndpointsToTrigger != null) {
final List<EndpointDTO> endpointsToTrigger = listOfEndpointsToTrigger;
for (final EndpointDTO endpoint : endpointsToTrigger) {
dispatchToSubscribersTask.submit(() -> {
dispatchToSubscribers(context, endpoint, bodyAsBytes);
dispatchToSubscribers(endpoint, bodyAsBytes);
});
}
// wait for tasks to complete for <number of end-points>*<timeout for connection>+<ramp-up time>
@@ -65,34 +94,36 @@ public class FixIngestionLambda implements RequestStreamHandler {
(endpointsToTrigger.size() * Configuration.TIMEOUT_IN_SECONDS_WHEN_DISPATCHING_TO_ENDPOINT) + 2,
TimeUnit.SECONDS);
} else {
context.getLogger().log("No endpoint has been configured for UUID " + dto.getDeviceUuid());
logger.info("No endpoint has been configured for Identifier " + deviceIdentifier.getStringRepresentation());
}
output.write(new Gson().toJson(AWSResponseWrapper.successResponseAsJson(dto.getDeviceUuid())).getBytes());
output.write(new Gson().toJson(AWSResponseWrapper.successResponseAsJson(deviceIdentifier.getStringRepresentation())).getBytes());
} catch (ParseException | JsonDeserializationException e) {
logger.log(Level.SEVERE, "Exception trying to deserialize JSON input: " + e.getMessage());
} catch (IOException e) {
context.getLogger().log(e.getMessage());
logger.log(Level.SEVERE, e.getMessage());
} finally {
try {
input.close();
} catch (IOException e) {
context.getLogger().log("Exception trying to close input: "+e.getMessage());
logger.log(Level.SEVERE, "Exception trying to close input: " + e.getMessage());
}
try {
output.close();
} catch (IOException e) {
context.getLogger().log("Exception trying to close output: "+e.getMessage());
logger.log(Level.SEVERE, "Exception trying to close output: " + e.getMessage());
}
}
}
private void dispatchToSubscribers(final Context context, final EndpointDTO endpoint, final byte[] jsonAsBytes) {
context.getLogger().log("Connecting to endpoint " + endpoint.getEndpointCallbackUrl()+" with ID "+endpoint.getEndpointUuid());
private void dispatchToSubscribers(final EndpointDTO endpoint, final byte[] jsonAsBytes) {
logger.info("Connecting to endpoint " + endpoint.getEndpointCallbackUrl()+" with ID "+endpoint.getEndpointUuid());
URL endpointUrl;
try {
endpointUrl = new URL(endpoint.getEndpointCallbackUrl());
try {
final HttpURLConnection connectionToEndpoint = (HttpURLConnection) endpointUrl.openConnection();
connectionToEndpoint.setRequestMethod("POST");
connectionToEndpoint.setRequestProperty("Content-Type", "application/json; utf-8");
connectionToEndpoint.setRequestProperty("Content-Type", "application/json; charset=utf-8");
connectionToEndpoint.setRequestProperty("Accept", "application/json");
connectionToEndpoint.setDoOutput(true);
connectionToEndpoint.addRequestProperty("Content-Length", String.valueOf(jsonAsBytes.length));
@@ -102,32 +133,28 @@ public class FixIngestionLambda implements RequestStreamHandler {
os.write(jsonAsBytes);
os.flush();
final int responseCode = connectionToEndpoint.getResponseCode(); // reading is important to actually issue the request
context.getLogger().log("Sent data "+new String(jsonAsBytes)+" to " + endpoint.getEndpointCallbackUrl()+" with response code "+responseCode);
logger.info("Sent data "+new String(jsonAsBytes)+" to " + endpoint.getEndpointCallbackUrl()+" with response code "+responseCode);
}
} catch (Exception ex) {
context.getLogger().log("Exception trying to send data to "+endpointUrl+": "+ex.getMessage());
logger.log(Level.SEVERE, "Exception trying to send data to "+endpointUrl+": "+ex.getMessage());
}
} catch (MalformedURLException e) {
context.getLogger().log("Malformed URL for end point "+endpoint.getEndpointCallbackUrl());
logger.log(Level.SEVERE, "Malformed URL for end point "+endpoint.getEndpointCallbackUrl());
}
}
private void storeFixFileToS3(final String deviceUuid, final byte[] jsonAsBytes, final LambdaLogger logger)
private void storeFixFileToS3(final DeviceIdentifier deviceIdentifier, final List<GPSFixMoving> newFixes)
throws IOException {
try (final S3Client s3Client = S3Client.builder().region(Configuration.S3_REGION).build()) {
final String destinationKey = getDestinationKey(deviceUuid);
final String dataAsString = newFixes.toString();
logger.info("Data to write: "+dataAsString);
for (GPSFixMoving fix: newFixes) {
final String destinationKey = s3FixStorageStructure.generateKeyForSingleFix(deviceIdentifier, fix.getTimePoint());
logger.info("Location: "+destinationKey);
final PutObjectRequest putObjectRequest = PutObjectRequest.builder().bucket(Configuration.S3_BUCKET_NAME)
.key(destinationKey).contentType("application/json").build();
s3Client.putObject(putObjectRequest, RequestBody.fromBytes(jsonAsBytes));
logger.log("Finished putting object into S3");
final JSONObject serializedFix = serializer.serialize(fix);
s3Client.putObject(putObjectRequest, RequestBody.fromString(serializedFix.toJSONString()));
}
}
private String getDestinationKey(final String deviceUuid) {
return getUuidSplitIntoS3Prefixes(deviceUuid) + "/" + LocalDateTime.now().toString() + ".json";
}
private String getUuidSplitIntoS3Prefixes(final String uuid) {
return String.join("/", uuid.split("-"));
logger.info("Finished putting object into S3");
}
}
java/com.sap.sailing.ingestion/src/main/java/com/sap/sailing/ingestion/S3FixStorageListRequest.java
+34
View File
@@ -0,0 +1,34 @@
package com.sap.sailing.ingestion;
/**
* This class defines a structure to facilitate the creation of a {@link ListObjectsV2Request} to the S3. The instance
* variables abstract the used basic structure of a {@link ListObjectsV2Request}. It defines a prefix, a key after which
* the response should start (keys on results are UTF-8 sorted) and the max amount of keys to give in one batch.
*
* @author Kevin Wiesner
*
*/
public class S3FixStorageListRequest {
private final String commonPrefix;
private final String keyStartAfter;
private final int maxKeys;
public S3FixStorageListRequest(String commonPrefix, String keyStartAfter, int maxKeys) {
super();
this.commonPrefix = commonPrefix;
this.keyStartAfter = keyStartAfter;
this.maxKeys = maxKeys;
}
public String getCommonPrefix() {
return commonPrefix;
}
public String getKeyStartAfter() {
return keyStartAfter;
}
public int getMaxKeys() {
return maxKeys;
}
}
java/com.sap.sailing.ingestion/src/main/java/com/sap/sailing/ingestion/S3FixStorageStructure.java
+247
View File
@@ -0,0 +1,247 @@
package com.sap.sailing.ingestion;
import java.util.Calendar;
import java.util.TimeZone;
import java.util.UUID;
import com.sap.sailing.domain.common.DeviceIdentifier;
import com.sap.sailing.domain.common.tracking.GPSFixMoving;
import com.sap.sailing.domain.racelogtracking.impl.SmartphoneUUIDIdentifierImpl;
import com.sap.sse.common.Duration;
import com.sap.sse.common.TimePoint;
import com.sap.sse.common.TimeRange;
import com.sap.sse.common.impl.MillisecondsDurationImpl;
import com.sap.sse.common.impl.TimeRangeImpl;
/**
* This class defines the organization structure of fixes on the S3 storage and is the foundation for all the AWS Lambda
* operations. The basic structure distinguishes single fixes which are inserted by the {@link FixIngestionLambda} and
* collection items which are created by the {@link FixCombinationLambda} by combining single fixes which are all in the
* same defined time range. It also facilitates the creation of list requests to obtain fixes of a requested time range.
* <br>
* <br>
* On S3, "files" are represented by keys and a defined delimiter can be used to imitate a folder structure to organize
* "files" hierarchically. The structure of all the keys on the S3 is the following:
* <ol>
* <li>Prefix (single fix or collection item)</li>
* <li>Device identifier of tracker</li>
* <li>Time of fix or collection</li>
* </ol>
*
* @author Kevin Wiesner
*
*/
public class S3FixStorageStructure {
private static final String SINGLE_FIX_PREFIX = "ingestion";
private static final String COLLECTION_PREFIX = "collection";
private static final String S3_DELIMITER = "/";
private static final Duration COLLECTION_DURATION = new MillisecondsDurationImpl(600_000);
private static final int SINGLE_FIX_BATCH_SIZE = 10;
/**
* Returns the basic prefix for a single fix.
*
* @return singleFixPrefix
*/
public String getSingleFixPrefix() {
return combineElementsToPrefix(SINGLE_FIX_PREFIX);
}
/**
* Generates a key for a single fix based on a given {@link DeviceIdentifier} and {@link TimePoint}.
*
* @param deviceIdentifier
* @param timePointOfSingleFix
* @return generatedKey
*/
public String generateKeyForSingleFix(final DeviceIdentifier deviceIdentifier,
final TimePoint timePointOfSingleFix) {
return generateS3KeyForFile(SINGLE_FIX_PREFIX, deviceIdentifier, timePointOfSingleFix);
}
/**
* Returns a boolean to determine whether a given {@link GPSFixMoving} object lies within the current collection time frame.
* This is useful to avoid to constantly add fixes to a single collection but to wait for a {@link TimeRange} of a
* collection to end and add all single fixes in one operation to a collection.
*
* @param fixToCheck
* @return isInCurrentCollectionFrame
*/
public Boolean isFixInCurrentCollectionFrame(final GPSFixMoving fixToCheck) {
final TimePoint fixTimePoint = fixToCheck.getTimePoint();
final TimeRange currentTimeRange = assignTimePointToCollectionTimeUnit(TimePoint.now());
return currentTimeRange.includes(fixTimePoint);
}
/**
* Creates a {@link S3FixStorageListRequest} object to facilitate the access to remaining single fixes in a defined
* {@link TimeRange}. By using the instance variables of the returning object, it is possible to create a
* {@link ListObjectsV2Request} to get the requested single fixes. The response to the created request would be structured
* in batches limited by the maxKeys property of the request, it is the task of the code using the request to end
* the process of accessing new batches when the desired {@link TimeRange} is covered by the single fixes received.
*
* @param deviceIdentifier
* @param timeRange
* @return listObjectRequestFacilitator
*/
public S3FixStorageListRequest getSingleFixesRequestForTimeRange(final DeviceIdentifier deviceIdentifier,
final TimeRange timeRange) {
final String keyStartAfter = generateKeyForSingleFix(deviceIdentifier, timeRange.from().minus(1));
final String keyToEnd = generateKeyForSingleFix(deviceIdentifier, timeRange.to());
final String commonPrefix = getCommonPrefix(keyStartAfter, keyToEnd);
return new S3FixStorageListRequest(commonPrefix, keyStartAfter, SINGLE_FIX_BATCH_SIZE);
}
/**
* Returns the basic prefix for a collection item.
*
* @return collectionPrefix
*/
public String getCollectionPrefix() {
return combineElementsToPrefix(COLLECTION_PREFIX);
}
/**
* Generates a key for a collection based on a given {@link DeviceIdentifier} and {@link TimeRange}.
*
* @param deviceIdentifier
* @param timeRangeOfFixCollection
* @return generatedKey
*/
public String generateKeyForCollection(final DeviceIdentifier deviceIdentifier,
final TimeRange timeRangeOfFixCollection) {
return generateS3KeyForFile(COLLECTION_PREFIX, deviceIdentifier, timeRangeOfFixCollection.from());
}
/**
* Maps a given fix to a collection {@link TimeRange}. Based on the defined COLLECTION_DURATION, it defines all {@link TimeRange}
* results beginning at timestamp(0).
*
* @param fixToAssign
* @return timeRangeForFixCollection
*/
public TimeRange assignFixToCollectionTimeUnit(final GPSFixMoving fixToAssign) {
return assignTimePointToCollectionTimeUnit(fixToAssign.getTimePoint());
}
/**
* Creates a {@link S3FixStorageListRequest} object to facilitate the access to the collection items in a defined {@link TimeRange}.
* By using the instance variables of the returning object, it is possible to create a {@link ListObjectsV2Request} to get
* the requested collection items. The response to the created request would be structured in one batch which
* contains all the relevant collection items. Is has to be considered that due to the fact that there could be a
* collection {@link TimeRange} without a collection item, it is possible that the batch would contain collection items
* which are out of the scope of the desired {@link TimeRange}.
*
* @param deviceIdentifier
* @param timeRange
* @return listObjectRequestFacilitator
*/
public S3FixStorageListRequest getCollectionRequestForTimeRange(final DeviceIdentifier deviceIdentifier,
final TimeRange timeRange) {
final TimeRange collectionAdjustedTimeRangeStart = assignTimePointToCollectionTimeUnit(
timeRange.from().minus(COLLECTION_DURATION));
final TimeRange collectionAdjustedTimeRangeEnd = assignTimePointToCollectionTimeUnit(timeRange.to());
final String keyStartAfter = generateKeyForCollection(deviceIdentifier, collectionAdjustedTimeRangeStart);
final String keyToEnd = generateKeyForCollection(deviceIdentifier, collectionAdjustedTimeRangeEnd);
final String commonPrefix = getCommonPrefix(keyStartAfter, keyToEnd);
final Duration requestDuration = collectionAdjustedTimeRangeStart.to()
.until(collectionAdjustedTimeRangeEnd.to());
final int maxKeys = (int) requestDuration.divide(COLLECTION_DURATION);
return new S3FixStorageListRequest(commonPrefix, keyStartAfter, maxKeys);
}
/**
* Extracts a {@link DeviceIdentifier} from a given key. The key has to follow the described structure of fixes on the
* storage. Currently, the {@link DeviceIdentifier} can only be a {@link UUID}.
*
* @param key
* @return deviceIdentifier
*/
public DeviceIdentifier getIdentifierFromKey(final String key) {
// TODO handle different device identifiers - improve handling
String deviceIdentifierString = "";
String[] keySplitAfterPrefix;
if (key.contains(COLLECTION_PREFIX)) {
keySplitAfterPrefix = key.split(COLLECTION_PREFIX);
} else {
keySplitAfterPrefix = key.split(SINGLE_FIX_PREFIX);
}
deviceIdentifierString = String.join("", keySplitAfterPrefix).split(S3_DELIMITER)[1];
final UUID deviceUuid = UUID.fromString(deviceIdentifierString);
final DeviceIdentifier deviceIdentifier = new SmartphoneUUIDIdentifierImpl(deviceUuid);
return deviceIdentifier;
}
/**
* Helper function to obtain the common prefix of two given keys.
*
* @param key1
* @param key2
* @return commonPrefix
*/
private String getCommonPrefix(final String key1, final String key2) {
final StringBuilder prefix = new StringBuilder();
int commonIndex = 0;
while (commonIndex != key1.length() && commonIndex != key2.length()
&& key1.charAt(commonIndex) == key2.charAt(commonIndex)) {
prefix.append(key1.charAt(commonIndex));
commonIndex += 1;
}
return prefix.toString();
}
/**
* Uses a given prefix, {@link DeviceIdentifier} and a {@link TimePoint} to create a key based on the determined storage structure.
*
* @param prefix
* @param deviceIdentifier
* @param timePoint
* @return generatedKey
*/
private String generateS3KeyForFile(final String prefix, final DeviceIdentifier deviceIdentifier,
final TimePoint timePoint) {
final Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
calendar.setTimeInMillis(timePoint.asMillis());
final String collectionName = String.format("%02d:%02d:%02d.%03d-UTC.json", calendar.get(Calendar.HOUR_OF_DAY),
calendar.get(Calendar.MINUTE), calendar.get(Calendar.SECOND), calendar.get(Calendar.MILLISECOND));
final String year = String.format("%04d", calendar.get(Calendar.YEAR));
final String month = String.format("%02d", calendar.get(Calendar.MONTH));
final String dayOfMonth = String.format("%02d", calendar.get(Calendar.DATE));
final String generatedKey = combineElementsToKey(prefix, deviceIdentifier.getStringRepresentation(), year,
month, dayOfMonth, collectionName);
return generatedKey;
}
/**
* Helper function to map a given {@link TimePoint} to a structured {@link TimeRange}.
*
* @param timePointToAssign
* @return mappedTimeRange
*/
private TimeRange assignTimePointToCollectionTimeUnit(final TimePoint timePointToAssign) {
final TimePoint rangeStart = timePointToAssign
.minus(timePointToAssign.asMillis() % COLLECTION_DURATION.asMillis());
final TimePoint rangeEnd = rangeStart.plus(COLLECTION_DURATION.minus(1));
return new TimeRangeImpl(rangeStart, rangeEnd);
}
/**
* Helper function to combine elements of a path to a key using the defined class delimiter.
*
* @param pathElements
* @return key
*/
private String combineElementsToKey(final String... pathElements) {
return String.join(S3_DELIMITER, pathElements);
}
/**
* Helper function to combine elements of a path to a prefix using the defined class delimiter.
*
* @param pathElements
* @return prefix
*/
private String combineElementsToPrefix(final String... pathElements) {
return combineElementsToKey(pathElements) + S3_DELIMITER;
}
}