src/main/java/org/onap/dcae/restapi/VesRestController.java

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * PROJECT
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright (C) 2018 Nokia. All rights reserved.s
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.dcae.restapi;
  23
  24 import static java.util.Optional.ofNullable;
  25 import static java.util.stream.StreamSupport.stream;
  26 import static org.springframework.http.ResponseEntity.ok;
  27
  28 import com.att.nsa.clock.SaClock;
  29 import com.att.nsa.logging.LoggingContext;
  30 import com.att.nsa.logging.log4j.EcompFields;
  31 import com.github.fge.jackson.JsonLoader;
  32 import com.github.fge.jsonschema.core.report.ProcessingReport;
  33 import com.github.fge.jsonschema.main.JsonSchema;
  34
  35 import java.util.UUID;
  36 import java.util.concurrent.LinkedBlockingQueue;
  37 import javax.servlet.http.HttpServletRequest;
  38
  39 import org.json.JSONArray;
  40 import org.json.JSONObject;
  41 import org.onap.dcae.ApplicationSettings;
  42 import org.onap.dcae.CollectorSchemas;
  43 import org.onap.dcae.commonFunction.VESLogger;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46 import org.springframework.beans.factory.annotation.Autowired;
  47 import org.springframework.beans.factory.annotation.Qualifier;
  48 import org.springframework.http.MediaType;
  49 import org.springframework.http.ResponseEntity;
  50 import org.springframework.web.bind.annotation.GetMapping;
  51 import org.springframework.web.bind.annotation.PostMapping;
  52 import org.springframework.web.bind.annotation.RequestBody;
  53 import org.springframework.web.bind.annotation.RestController;
  54
  55 @RestController
  56 public class VesRestController {
  57
  58     private static final Logger LOG = LoggerFactory.getLogger(VesRestController.class);
  59
  60     private static final String FALLBACK_VES_VERSION = "v5";
  61
  62     @Autowired
  63     private ApplicationSettings collectorProperties;
  64
  65     @Autowired
  66     private CollectorSchemas schemas;
  67
  68     @Autowired
  69     @Qualifier("metriclog")
  70     private Logger metriclog;
  71
  72     @Autowired
  73     @Qualifier("incomingRequestsLogger")
  74     private Logger incomingRequestsLogger;
  75
  76     @Autowired
  77     @Qualifier("errorLog")
  78     private Logger errorLog;
  79
  80     private LinkedBlockingQueue<JSONObject> inputQueue;
  81     private String version;
  82
  83     @Autowired
  84     VesRestController(@Qualifier("incomingRequestsLogger") Logger incomingRequestsLogger,
  85                       @Qualifier("inputQueue") LinkedBlockingQueue<JSONObject> inputQueue) {
  86         this.incomingRequestsLogger = incomingRequestsLogger;
  87         this.inputQueue = inputQueue;
  88     }
  89
  90     @GetMapping("/")
  91     String mainPage() {
  92         return "Welcome to VESCollector";
  93     }
  94
  95     //refactor in next iteration
  96     @PostMapping(value = {"/eventListener/v1",
  97             "/eventListener/v1/eventBatch",
  98             "/eventListener/v2",
  99             "/eventListener/v2/eventBatch",
 100             "/eventListener/v3",
 101             "/eventListener/v3/eventBatch",
 102             "/eventListener/v4",
 103             "/eventListener/v4/eventBatch",
 104             "/eventListener/v5",
 105             "/eventListener/v5/eventBatch"}, consumes = "application/json")
 106     ResponseEntity<String> receiveEvent(@RequestBody String jsonPayload, HttpServletRequest httpServletRequest) {
 107         String request = httpServletRequest.getRequestURI();
 108         extractVersion(request);
 109
 110         JSONObject jsonObject;
 111         try {
 112             jsonObject = new JSONObject(jsonPayload);
 113         } catch (Exception e) {
 114             return ResponseEntity.badRequest().body(ApiException.INVALID_JSON_INPUT.toJSON().toString());
 115         }
 116
 117         String uuid = setUpECOMPLoggingForRequest();
 118         incomingRequestsLogger.info(String.format(
 119                 "Received a VESEvent '%s', marked with unique identifier '%s', on api version '%s', from host: '%s'",
 120                 jsonObject, uuid, version, httpServletRequest.getRemoteHost()));
 121
 122         if (collectorProperties.jsonSchemaValidationEnabled()) {
 123             if (isBatchRequest(request) && (jsonObject.has("eventList") && (!jsonObject.has("event")))) {
 124                 if (!conformsToSchema(jsonObject, version)) {
 125                     return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED);
 126                 }
 127             } else if (!isBatchRequest(request) && (!jsonObject.has("eventList") && (jsonObject.has("event")))) {
 128                 if (!conformsToSchema(jsonObject, version)) {
 129                     return errorResponse(ApiException.SCHEMA_VALIDATION_FAILED);
 130                 }
 131             } else {
 132                 return errorResponse(ApiException.INVALID_JSON_INPUT);
 133             }
 134         }
 135
 136         JSONArray commonlyFormatted = convertToJSONArrayCommonFormat(jsonObject, request, uuid, version);
 137
 138         if (!putEventsOnProcessingQueue(commonlyFormatted)) {
 139             errorLog.error("EVENT_RECEIPT_FAILURE: QueueFull " + ApiException.NO_SERVER_RESOURCES);
 140             return errorResponse(ApiException.NO_SERVER_RESOURCES);
 141         }
 142         return ok().contentType(MediaType.APPLICATION_JSON).body("Message Accepted");
 143     }
 144
 145     private void extractVersion(String httpServletRequest) {
 146         version = httpServletRequest.split("/")[2];
 147     }
 148
 149     private ResponseEntity<String> errorResponse(ApiException noServerResources) {
 150         return ResponseEntity.status(noServerResources.httpStatusCode)
 151                 .body(noServerResources.toJSON().toString());
 152     }
 153
 154     private boolean putEventsOnProcessingQueue(JSONArray arrayOfEvents) {
 155         for (int i = 0; i < arrayOfEvents.length(); i++) {
 156             metriclog.info("EVENT_PUBLISH_START");
 157             if (!inputQueue.offer((JSONObject) arrayOfEvents.get(i))) {
 158                 return false;
 159             }
 160         }
 161         LOG.debug("CommonStartup.handleEvents:EVENTS has been published successfully!");
 162         metriclog.info("EVENT_PUBLISH_END");
 163         return true;
 164     }
 165
 166     private boolean conformsToSchema(JSONObject payload, String version) {
 167         try {
 168             JsonSchema schema = ofNullable(schemas.getJSONSchemasMap(version).get(version))
 169                     .orElse(schemas.getJSONSchemasMap(version).get(FALLBACK_VES_VERSION));
 170             ProcessingReport report = schema.validate(JsonLoader.fromString(payload.toString()));
 171             if (!report.isSuccess()) {
 172                 LOG.warn("Schema validation failed for event: " + payload);
 173                 stream(report.spliterator(), false).forEach(e -> LOG.warn(e.getMessage()));
 174                 return false;
 175             }
 176             return report.isSuccess();
 177         } catch (Exception e) {
 178             throw new RuntimeException("Unable to validate against schema", e);
 179         }
 180     }
 181
 182     private static JSONArray convertToJSONArrayCommonFormat(JSONObject jsonObject, String request,
 183                                                             String uuid, String version) {
 184         JSONArray asArrayEvents = new JSONArray();
 185         String vesUniqueIdKey = "VESuniqueId";
 186         String vesVersionKey = "VESversion";
 187         if (isBatchRequest(request)) {
 188             JSONArray events = jsonObject.getJSONArray("eventList");
 189             for (int i = 0; i < events.length(); i++) {
 190                 JSONObject event = new JSONObject().put("event", events.getJSONObject(i));
 191                 event.put(vesUniqueIdKey, uuid + "-" + i);
 192                 event.put(vesVersionKey, version);
 193                 asArrayEvents.put(event);
 194             }
 195         } else {
 196             jsonObject.put(vesUniqueIdKey, uuid);
 197             jsonObject.put(vesVersionKey, version);
 198             asArrayEvents = new JSONArray().put(jsonObject);
 199         }
 200         return asArrayEvents;
 201     }
 202
 203     private static String setUpECOMPLoggingForRequest() {
 204         final UUID uuid = UUID.randomUUID();
 205         LoggingContext localLC = VESLogger.getLoggingContextForThread(uuid);
 206         localLC.put(EcompFields.kBeginTimestampMs, SaClock.now());
 207         return uuid.toString();
 208     }
 209
 210     private static boolean isBatchRequest(String request) {
 211         return request.contains("eventBatch");
 212     }
 213 }