diff options
Diffstat (limited to 'src/main/java/org/onap/dcae/restapi/VesRestController.java')
| -rw-r--r-- | src/main/java/org/onap/dcae/restapi/VesRestController.java | 301 |
1 files changed, 157 insertions, 144 deletions
diff --git a/src/main/java/org/onap/dcae/restapi/VesRestController.java b/src/main/java/org/onap/dcae/restapi/VesRestController.java index 92e8d004..58b52765 100644 --- a/src/main/java/org/onap/dcae/restapi/VesRestController.java +++ b/src/main/java/org/onap/dcae/restapi/VesRestController.java @@ -55,159 +55,172 @@ import org.springframework.web.bind.annotation.RestController; @RestController public class VesRestController { - private static final Logger LOG = LoggerFactory.getLogger(VesRestController.class); - - private static final String FALLBACK_VES_VERSION = "v5"; - - @Autowired - private ApplicationSettings collectorProperties; - - @Autowired - private CollectorSchemas schemas; - - @Autowired - @Qualifier("metriclog") - private Logger metriclog; - - @Autowired - @Qualifier("incomingRequestsLogger") - private Logger incomingRequestsLogger; - - @Autowired - @Qualifier("errorLog") - private Logger errorLog; - - private LinkedBlockingQueue<JSONObject> inputQueue; - private String version; - - @Autowired - VesRestController(@Qualifier("incomingRequestsLogger") Logger incomingRequestsLogger, - @Qualifier("inputQueue") LinkedBlockingQueue<JSONObject> inputQueue) { - this.incomingRequestsLogger = incomingRequestsLogger; - this.inputQueue = inputQueue; - } - - @GetMapping("/") - String mainPage() { - return "Welcome to VESCollector"; + private static final Logger LOG = LoggerFactory.getLogger(VesRestController.class); + + private static final String FALLBACK_VES_VERSION = "v5"; + + @Autowired private ApplicationSettings collectorProperties; + + @Autowired private CollectorSchemas schemas; + + @Autowired + @Qualifier("metriclog") + private Logger metriclog; + + @Autowired + @Qualifier("incomingRequestsLogger") + private Logger incomingRequestsLogger; + + @Autowired + @Qualifier("errorLog") + private Logger errorLog; + + private LinkedBlockingQueue<JSONObject> inputQueue; + private String version; + + @Autowired + VesRestController( + @Qualifier("incomingRequestsLogger") Logger incomingRequestsLogger, + @Qualifier("inputQueue") LinkedBlockingQueue<JSONObject> inputQueue) { + this.incomingRequestsLogger = incomingRequestsLogger; + this.inputQueue = inputQueue; + } + + @GetMapping("/") + String mainPage() { + return "Welcome to VESCollector"; + } + + // refactor in next iteration + @PostMapping( + value = { + "/eventListener/v1", + "/eventListener/v1/eventBatch", + "/eventListener/v2", + "/eventListener/v2/eventBatch", + "/eventListener/v3", + "/eventListener/v3/eventBatch", + "/eventListener/v4", + "/eventListener/v4/eventBatch", + "/eventListener/v5", + "/eventListener/v5/eventBatch", + "/eventListener/v7", + "/eventListener/v7/eventBatch" + }, + consumes = "application/json") + ResponseEntity<String> receiveEvent( + @RequestBody String jsonPayload, HttpServletRequest httpServletRequest) { + String request = httpServletRequest.getRequestURI(); + extractVersion(request); + + JSONObject jsonObject; + try { + jsonObject = new JSONObject(jsonPayload); + } catch (Exception e) { + return ResponseEntity.badRequest().body(ApiException.INVALID_JSON_INPUT.toJSON().toString()); } - //refactor in next iteration - @PostMapping(value = {"/eventListener/v1", - "/eventListener/v1/eventBatch", - "/eventListener/v2", - "/eventListener/v2/eventBatch", - "/eventListener/v3", - "/eventListener/v3/eventBatch", - "/eventListener/v4", - "/eventListener/v4/eventBatch", - "/eventListener/v5", - "/eventListener/v5/eventBatch"}, consumes = "application/json") - ResponseEntity<String> receiveEvent(@RequestBody String jsonPayload, HttpServletRequest httpServletRequest) { - String request = httpServletRequest.getRequestURI(); - extractVersion(request); - - JSONObject jsonObject; - try { - jsonObject = new JSONObject(jsonPayload); - } catch (Exception e) { - return ResponseEntity.badRequest().body(ApiException.INVALID_JSON_INPUT.toJSON().toString()); - } + String uuid = setUpECOMPLoggingForRequest(); + incomingRequestsLogger.info( + String.format( + "Received a VESEvent '%s', marked with unique identifier '%s', on api version '%s', from host: '%s'", + jsonObject, uuid, version, httpServletRequest.getRemoteHost())); - String uuid = setUpECOMPLoggingForRequest(); - incomingRequestsLogger.info(String.format( - "Received a VESEvent '%s', marked with unique identifier '%s', on api version '%s', from host: '%s'", - jsonObject, uuid, version, httpServletRequest.getRemoteHost())); - - if (collectorProperties.jsonSchemaValidationEnabled()) { - if (isBatchRequest(request) && (jsonObject.has("eventList") && (!jsonObject.has("event")))) { - if (!conformsToSchema(jsonObject, version)) { - return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED); - } - } else if (!isBatchRequest(request) && (!jsonObject.has("eventList") && (jsonObject.has("event")))) { - if (!conformsToSchema(jsonObject, version)) { - return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED); - } - } else { - return errorResponse(ApiException.INVALID_JSON_INPUT); - } + if (collectorProperties.jsonSchemaValidationEnabled()) { + if (isBatchRequest(request) && (jsonObject.has("eventList") && (!jsonObject.has("event")))) { + if (!conformsToSchema(jsonObject, version)) { + return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED); } - - JSONArray commonlyFormatted = convertToJSONArrayCommonFormat(jsonObject, request, uuid, version); - - if (!putEventsOnProcessingQueue(commonlyFormatted)) { - errorLog.error("EVENT_RECEIPT_FAILURE: QueueFull " + ApiException.NO_SERVER_RESOURCES); - return errorResponse(ApiException.NO_SERVER_RESOURCES); + } else if (!isBatchRequest(request) + && (!jsonObject.has("eventList") && (jsonObject.has("event")))) { + if (!conformsToSchema(jsonObject, version)) { + return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED); } - return ok().contentType(MediaType.APPLICATION_JSON).body("Message Accepted"); - } - - private void extractVersion(String httpServletRequest) { - version = httpServletRequest.split("/")[2]; + } else { + return errorResponse(ApiException.INVALID_JSON_INPUT); + } } - private ResponseEntity<String> errorResponse(ApiException noServerResources) { - return ResponseEntity.status(noServerResources.httpStatusCode) - .body(noServerResources.toJSON().toString()); - } - - private boolean putEventsOnProcessingQueue(JSONArray arrayOfEvents) { - for (int i = 0; i < arrayOfEvents.length(); i++) { - metriclog.info("EVENT_PUBLISH_START"); - if (!inputQueue.offer((JSONObject) arrayOfEvents.get(i))) { - return false; - } - } - LOG.debug("CommonStartup.handleEvents:EVENTS has been published successfully!"); - metriclog.info("EVENT_PUBLISH_END"); - return true; - } + JSONArray commonlyFormatted = + convertToJSONArrayCommonFormat(jsonObject, request, uuid, version); - private boolean conformsToSchema(JSONObject payload, String version) { - try { - JsonSchema schema = ofNullable(schemas.getJSONSchemasMap(version).get(version)) - .orElse(schemas.getJSONSchemasMap(version).get(FALLBACK_VES_VERSION)); - ProcessingReport report = schema.validate(JsonLoader.fromString(payload.toString())); - if (!report.isSuccess()) { - LOG.warn("Schema validation failed for event: " + payload); - stream(report.spliterator(), false).forEach(e -> LOG.warn(e.getMessage())); - return false; - } - return report.isSuccess(); - } catch (Exception e) { - throw new RuntimeException("Unable to validate against schema", e); - } + if (!putEventsOnProcessingQueue(commonlyFormatted)) { + errorLog.error("EVENT_RECEIPT_FAILURE: QueueFull " + ApiException.NO_SERVER_RESOURCES); + return errorResponse(ApiException.NO_SERVER_RESOURCES); } - - private static JSONArray convertToJSONArrayCommonFormat(JSONObject jsonObject, String request, - String uuid, String version) { - JSONArray asArrayEvents = new JSONArray(); - String vesUniqueIdKey = "VESuniqueId"; - String vesVersionKey = "VESversion"; - if (isBatchRequest(request)) { - JSONArray events = jsonObject.getJSONArray("eventList"); - for (int i = 0; i < events.length(); i++) { - JSONObject event = new JSONObject().put("event", events.getJSONObject(i)); - event.put(vesUniqueIdKey, uuid + "-" + i); - event.put(vesVersionKey, version); - asArrayEvents.put(event); - } - } else { - jsonObject.put(vesUniqueIdKey, uuid); - jsonObject.put(vesVersionKey, version); - asArrayEvents = new JSONArray().put(jsonObject); - } - return asArrayEvents; + // HttpStatus.SC_NO_CONTENT + return org.springframework.http.ResponseEntity.accepted() + .contentType(MediaType.APPLICATION_JSON) + .body("Accepted"); + } + + private void extractVersion(String httpServletRequest) { + version = httpServletRequest.split("/")[2]; + } + + private ResponseEntity<String> errorResponse(ApiException noServerResources) { + return ResponseEntity.status(noServerResources.httpStatusCode) + .body(noServerResources.toJSON().toString()); + } + + private boolean putEventsOnProcessingQueue(JSONArray arrayOfEvents) { + for (int i = 0; i < arrayOfEvents.length(); i++) { + metriclog.info("EVENT_PUBLISH_START"); + if (!inputQueue.offer((JSONObject) arrayOfEvents.get(i))) { + return false; + } } - - private static String setUpECOMPLoggingForRequest() { - final UUID uuid = UUID.randomUUID(); - LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid); - localLC.put(EcompFields.kBeginTimestampMs, SaClock.now()); - return uuid.toString(); + LOG.debug("CommonStartup.handleEvents:EVENTS has been published successfully!"); + metriclog.info("EVENT_PUBLISH_END"); + return true; + } + + private boolean conformsToSchema(JSONObject payload, String version) { + try { + JsonSchema schema = + ofNullable(schemas.getJSONSchemasMap(version).get(version)) + .orElse(schemas.getJSONSchemasMap(version).get(FALLBACK_VES_VERSION)); + ProcessingReport report = schema.validate(JsonLoader.fromString(payload.toString())); + if (!report.isSuccess()) { + LOG.warn("Schema validation failed for event: " + payload); + stream(report.spliterator(), false).forEach(e -> LOG.warn(e.getMessage())); + return false; + } + return report.isSuccess(); + } catch (Exception e) { + throw new RuntimeException("Unable to validate against schema", e); } - - private static boolean isBatchRequest(String request) { - return request.contains("eventBatch"); + } + + private static JSONArray convertToJSONArrayCommonFormat( + JSONObject jsonObject, String request, String uuid, String version) { + JSONArray asArrayEvents = new JSONArray(); + String vesUniqueIdKey = "VESuniqueId"; + String vesVersionKey = "VESversion"; + if (isBatchRequest(request)) { + JSONArray events = jsonObject.getJSONArray("eventList"); + for (int i = 0; i < events.length(); i++) { + JSONObject event = new JSONObject().put("event", events.getJSONObject(i)); + event.put(vesUniqueIdKey, uuid + "-" + i); + event.put(vesVersionKey, version); + asArrayEvents.put(event); + } + } else { + jsonObject.put(vesUniqueIdKey, uuid); + jsonObject.put(vesVersionKey, version); + asArrayEvents = new JSONArray().put(jsonObject); } -}
\ No newline at end of file + return asArrayEvents; + } + + private static String setUpECOMPLoggingForRequest() { + final UUID uuid = UUID.randomUUID(); + LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid); + localLC.put(EcompFields.kBeginTimestampMs, SaClock.now()); + return uuid.toString(); + } + + private static boolean isBatchRequest(String request) { + return request.contains("eventBatch"); + } +} |