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 | 297 |
1 files changed, 137 insertions, 160 deletions
diff --git a/src/main/java/org/onap/dcae/restapi/VesRestController.java b/src/main/java/org/onap/dcae/restapi/VesRestController.java index 58b52765..21fa685c 100644 --- a/src/main/java/org/onap/dcae/restapi/VesRestController.java +++ b/src/main/java/org/onap/dcae/restapi/VesRestController.java @@ -21,9 +21,8 @@ package org.onap.dcae.restapi; -import static java.util.Optional.ofNullable; import static java.util.stream.StreamSupport.stream; -import static org.springframework.http.ResponseEntity.ok; +import static org.springframework.http.ResponseEntity.accepted; import com.att.nsa.clock.SaClock; import com.att.nsa.logging.LoggingContext; @@ -39,7 +38,6 @@ import javax.servlet.http.HttpServletRequest; import org.json.JSONArray; import org.json.JSONObject; import org.onap.dcae.ApplicationSettings; -import org.onap.dcae.CollectorSchemas; import org.onap.dcae.commonFunction.VESLogger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,173 +52,152 @@ import org.springframework.web.bind.annotation.RestController; @RestController public class VesRestController { + private static final Logger log = LoggerFactory.getLogger(VesRestController.class); + + private final ApplicationSettings applicationSettings; + private final LinkedBlockingQueue<JSONObject> inputQueue; + + private final Logger metricsLog; + private final Logger errorLog; + private final Logger incomingRequestsLogger; + + @Autowired + VesRestController(ApplicationSettings applicationSettings, + @Qualifier("metricsLog") Logger metricsLog, + @Qualifier("errorLog") Logger errorLog, + @Qualifier("incomingRequestsLogger") Logger incomingRequestsLogger, + @Qualifier("inputQueue") LinkedBlockingQueue<JSONObject> inputQueue) { + this.applicationSettings = applicationSettings; + this.metricsLog = metricsLog; + this.errorLog = errorLog; + this.incomingRequestsLogger = incomingRequestsLogger; + this.inputQueue = inputQueue; + } - 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()); + @GetMapping("/") + String mainPage() { + return "Welcome to VESCollector"; } - 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())); + //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(); + String version = extractVersion(request); + + JSONObject jsonObject; + try { + jsonObject = new JSONObject(jsonPayload); + } catch (Exception e) { + return ResponseEntity.badRequest().body(ApiException.INVALID_JSON_INPUT.toJSON().toString()); + } - if (collectorProperties.jsonSchemaValidationEnabled()) { - if (isBatchRequest(request) && (jsonObject.has("eventList") && (!jsonObject.has("event")))) { - if (!conformsToSchema(jsonObject, version)) { - return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED); + 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 (applicationSettings.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); + } } - } else 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 { - return errorResponse(ApiException.INVALID_JSON_INPUT); - } + return accepted() + .contentType(MediaType.APPLICATION_JSON) + .body("Accepted"); } - JSONArray commonlyFormatted = - convertToJSONArrayCommonFormat(jsonObject, request, uuid, version); + private String extractVersion(String httpServletRequest) { + return httpServletRequest.split("/")[2]; + } - if (!putEventsOnProcessingQueue(commonlyFormatted)) { - errorLog.error("EVENT_RECEIPT_FAILURE: QueueFull " + ApiException.NO_SERVER_RESOURCES); - return errorResponse(ApiException.NO_SERVER_RESOURCES); + private ResponseEntity<String> errorResponse(ApiException noServerResources) { + return ResponseEntity.status(noServerResources.httpStatusCode) + .body(noServerResources.toJSON().toString()); } - // 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 boolean putEventsOnProcessingQueue(JSONArray arrayOfEvents) { + for (int i = 0; i < arrayOfEvents.length(); i++) { + metricsLog.info("EVENT_PUBLISH_START"); + if (!inputQueue.offer((JSONObject) arrayOfEvents.get(i))) { + return false; + } + } + log.debug("CommonStartup.handleEvents:EVENTS has been published successfully!"); + metricsLog.info("EVENT_PUBLISH_END"); + return true; } - 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 boolean conformsToSchema(JSONObject payload, String version) { + try { + JsonSchema schema = applicationSettings.jsonSchema(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 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; } - } - - 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); + + 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"); } - 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"); - } -} +}
\ No newline at end of file |