src/main/java/org/onap/crud/service/CrudAsyncGraphDataService.java

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * org.onap.aai
   4  * ================================================================================
   5  * Copyright © 2017-2018 AT&T Intellectual Property. All rights reserved.
   6  * Copyright © 2017-2018 Amdocs
   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 package org.onap.crud.service;
  22
  23 import java.io.IOException;
  24 import java.security.NoSuchAlgorithmException;
  25 import java.text.SimpleDateFormat;
  26 import java.util.HashMap;
  27 import java.util.Timer;
  28 import java.util.concurrent.Callable;
  29 import java.util.concurrent.CountDownLatch;
  30 import java.util.concurrent.ExecutionException;
  31 import java.util.concurrent.ExecutorService;
  32 import java.util.concurrent.Executors;
  33 import java.util.concurrent.Future;
  34 import java.util.concurrent.TimeUnit;
  35 import java.util.concurrent.TimeoutException;
  36 import javax.annotation.PreDestroy;
  37 import javax.ws.rs.core.EntityTag;
  38 import javax.ws.rs.core.Response.Status;
  39 import org.apache.commons.lang3.tuple.ImmutablePair;
  40 import org.onap.aai.cl.api.LogFields;
  41 import org.onap.aai.cl.api.Logger;
  42 import org.onap.aai.cl.eelf.LoggerFactory;
  43 import org.onap.aai.cl.mdc.MdcContext;
  44 import org.onap.aai.cl.mdc.MdcOverride;
  45 import org.onap.aai.event.api.EventConsumer;
  46 import org.onap.aai.event.api.EventPublisher;
  47 import org.onap.aai.restclient.client.OperationResult;
  48 import org.onap.crud.dao.GraphDao;
  49 import org.onap.crud.entity.Edge;
  50 import org.onap.crud.entity.Vertex;
  51 import org.onap.crud.event.GraphEvent;
  52 import org.onap.crud.event.GraphEvent.GraphEventOperation;
  53 import org.onap.crud.event.GraphEventEdge;
  54 import org.onap.crud.event.GraphEventVertex;
  55 import org.onap.crud.event.envelope.GraphEventEnvelope;
  56 import org.onap.crud.event.response.GraphEventResponseHandler;
  57 import org.onap.crud.exception.CrudException;
  58 import org.onap.crud.logging.CrudServiceMsgs;
  59 import org.onap.crud.util.CrudProperties;
  60 import org.onap.crud.util.CrudServiceConstants;
  61 import org.onap.crud.util.etag.EtagGenerator;
  62 import org.onap.schema.OxmModelValidator;
  63 import org.onap.schema.RelationshipSchemaValidator;
  64
  65 public class CrudAsyncGraphDataService extends AbstractGraphDataService {
  66
  67     private static Integer requestTimeOut;
  68
  69     private EventPublisher asyncRequestPublisher;
  70
  71     private Timer timer;
  72
  73     public static final Integer DEFAULT_REQUEST_TIMEOUT = 30000;
  74     private static final Integer DEFAULT_ASYNC_RESPONSE_PROCESS_POLL_INTERVAL = 1000;
  75
  76     private static Logger logger = LoggerFactory.getInstance().getLogger(CrudAsyncGraphDataService.class.getName());
  77     private static Logger metricsLogger =
  78             LoggerFactory.getInstance().getMetricsLogger(CrudAsyncGraphDataService.class.getName());
  79     private static LogFields okFields = new LogFields();
  80     private EtagGenerator etagGenerator;
  81
  82     static {
  83         okFields.setField(Status.OK, Status.OK.toString());
  84     }
  85
  86     private GraphEventResponseHandler responseHandler = new GraphEventResponseHandler();
  87
  88     public static Integer getRequestTimeOut() {
  89         return requestTimeOut;
  90     }
  91
  92     public CrudAsyncGraphDataService(GraphDao dao, EventPublisher asyncRequestPublisher,
  93             EventConsumer asyncResponseConsumer) throws CrudException, NoSuchAlgorithmException {
  94         this(dao, dao, asyncRequestPublisher, asyncResponseConsumer);
  95     }
  96
  97     public CrudAsyncGraphDataService(GraphDao dao, GraphDao daoForGet, EventPublisher asyncRequestPublisher,
  98             EventConsumer asyncResponseConsumer) throws CrudException, NoSuchAlgorithmException {
  99
 100         super();
 101         this.dao = dao;
 102         this.daoForGet = daoForGet;
 103
 104         requestTimeOut = DEFAULT_REQUEST_TIMEOUT;
 105         try {
 106             requestTimeOut = Integer.parseInt(CrudProperties.get(CrudServiceConstants.CRD_ASYNC_REQUEST_TIMEOUT));
 107         } catch (NumberFormatException ex) {
 108             // Leave it as the default
 109         }
 110
 111         Integer responsePollInterval = DEFAULT_ASYNC_RESPONSE_PROCESS_POLL_INTERVAL;
 112         try {
 113             responsePollInterval =
 114                     Integer.parseInt(CrudProperties.get(CrudServiceConstants.CRD_ASYNC_RESPONSE_PROCESS_POLL_INTERVAL));
 115         } catch (Exception ex) {
 116             logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR, "Unable to parse "
 117                     + CrudServiceConstants.CRD_ASYNC_RESPONSE_PROCESS_POLL_INTERVAL + " error: " + ex.getMessage());
 118         }
 119
 120         // Start the Response Consumer timer
 121         CrudAsyncResponseConsumer crudAsyncResponseConsumer = new CrudAsyncResponseConsumer(asyncResponseConsumer);
 122         timer = new Timer("crudAsyncResponseConsumer-1");
 123         timer.schedule(crudAsyncResponseConsumer, responsePollInterval, responsePollInterval);
 124
 125         this.asyncRequestPublisher = asyncRequestPublisher;
 126         this.etagGenerator = new EtagGenerator();
 127
 128         logger.info(CrudServiceMsgs.ASYNC_DATA_SERVICE_INFO, "CrudAsyncGraphDataService initialized SUCCESSFULLY!");
 129     }
 130
 131     public class CollectGraphResponse implements Callable<GraphEventEnvelope> {
 132         private volatile GraphEventEnvelope graphEventEnvelope;
 133         private volatile CountDownLatch latch = new CountDownLatch(1);
 134
 135         @Override
 136         public GraphEventEnvelope call() throws TimeoutException {
 137             try {
 138                 // Wait until graphEvent is available
 139                 latch.await(CrudAsyncGraphDataService.getRequestTimeOut(), TimeUnit.MILLISECONDS);
 140             } catch (InterruptedException e) {
 141                 latch.countDown();
 142                 if (this.graphEventEnvelope != null) {
 143                     return this.graphEventEnvelope;
 144                 } else {
 145                     throw new TimeoutException();
 146                 }
 147             }
 148             return this.graphEventEnvelope;
 149         }
 150
 151         public void populateGraphEventEnvelope(GraphEventEnvelope eventEnvelope) {
 152             this.graphEventEnvelope = eventEnvelope;
 153             latch.countDown();
 154         }
 155     }
 156
 157     private GraphEventEnvelope sendAndWait(GraphEvent event) throws CrudException {
 158
 159         long startTimeInMs = System.currentTimeMillis();
 160         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSXXX");
 161         MdcOverride override = new MdcOverride();
 162         override.addAttribute(MdcContext.MDC_START_TIME, formatter.format(startTimeInMs));
 163
 164         String eventEnvelopeJson = new GraphEventEnvelope(event).toJson();
 165
 166         // publish to request queue
 167         try {
 168             asyncRequestPublisher.sendSync(eventEnvelopeJson);
 169         } catch (Exception e) {
 170             throw new CrudException(
 171                     "Error publishing request " + event.getTransactionId() + "  Cause: " + e.getMessage(),
 172                     Status.INTERNAL_SERVER_ERROR);
 173         }
 174
 175         logger.debug(CrudServiceMsgs.ASYNC_DATA_SERVICE_INFO, "Event Sent =" + eventEnvelopeJson);
 176
 177         logger.info(CrudServiceMsgs.ASYNC_DATA_SERVICE_INFO,
 178                 "Event submitted of type: " + event.getObjectType() + " with key: " + event.getObjectKey()
 179                         + " , transaction-id: " + event.getTransactionId() + " , operation: "
 180                         + event.getOperation().toString());
 181
 182         ExecutorService executor =
 183                 Executors.newSingleThreadExecutor(new CrudThreadFactory("TX-" + event.getTransactionId()));
 184         CollectGraphResponse collector = new CollectGraphResponse();
 185         CrudAsyncGraphEventCache.put(event.getTransactionId(), collector);
 186         GraphEventEnvelope response;
 187         Future<GraphEventEnvelope> future = executor.submit(collector);
 188         try {
 189             response = future.get(requestTimeOut, TimeUnit.MILLISECONDS);
 190
 191         } catch (InterruptedException | ExecutionException | TimeoutException e) {
 192             CrudAsyncGraphEventCache.invalidate(event.getTransactionId());
 193             logger.error(CrudServiceMsgs.ASYNC_DATA_SERVICE_ERROR,
 194                     "Request timed out for transactionId: " + event.getTransactionId());
 195             future.cancel(true);
 196             throw new CrudException("Timed out , transactionId: " + event.getTransactionId() + " , operation: "
 197                     + event.getOperation().toString(), Status.INTERNAL_SERVER_ERROR);
 198         } finally {
 199             // Kill the thread as the work is completed
 200             executor.shutdownNow();
 201         }
 202         metricsLogger.info(CrudServiceMsgs.ASYNC_DATA_SERVICE_INFO, okFields, override,
 203                 "Total elapsed time for operation: " + event.getOperation().toString() + " , transactionId: "
 204                         + event.getTransactionId() + " is " + Long.toString(System.currentTimeMillis() - startTimeInMs)
 205                         + " ms");
 206         return response;
 207     }
 208
 209     @Override
 210     public ImmutablePair<EntityTag, String> addVertex(String version, String type, VertexPayload payload)
 211             throws CrudException {
 212         // Validate the incoming payload
 213         Vertex vertex = OxmModelValidator.validateIncomingUpsertPayload(null, version, type, payload.getProperties());
 214         vertex.getProperties().put(org.onap.schema.OxmModelValidator.Metadata.NODE_TYPE.propertyName(), type);
 215         // Create graph request event
 216         GraphEvent event = GraphEvent.builder(GraphEventOperation.CREATE)
 217                 .vertex(GraphEventVertex.fromVertex(vertex, version)).build();
 218
 219         GraphEventEnvelope response = sendAndWait(event);
 220
 221         EntityTag entityTag;
 222         try {
 223             entityTag = new EntityTag(etagGenerator.computeHashForVertex(response.getBody().getVertex()));
 224         } catch (IOException e) {
 225             throw new CrudException(e);
 226         }
 227         String responsePayload = responseHandler.handleVertexResponse(version, event, response);
 228
 229         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 230     }
 231
 232     @Override
 233     public ImmutablePair<EntityTag, String> addEdge(String version, String type, EdgePayload payload)
 234             throws CrudException {
 235         Edge edge = RelationshipSchemaValidator.validateIncomingAddPayload(version, type, payload);
 236         // Create graph request event
 237         GraphEvent event =
 238                 GraphEvent.builder(GraphEventOperation.CREATE).edge(GraphEventEdge.fromEdge(edge, version)).build();
 239
 240         GraphEventEnvelope response = sendAndWait(event);
 241
 242         EntityTag entityTag;
 243         try {
 244             entityTag = new EntityTag(etagGenerator.computeHashForEdge(response.getBody().getEdge()));
 245         } catch (IOException e) {
 246             throw new CrudException(e);
 247         }
 248         String responsePayload = responseHandler.handleEdgeResponse(version, event, response);
 249
 250         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 251     }
 252
 253     @Override
 254     public ImmutablePair<EntityTag, String> updateVertex(String version, String id, String type, VertexPayload payload)
 255             throws CrudException {
 256         Vertex vertex = OxmModelValidator.validateIncomingUpsertPayload(id, version, type, payload.getProperties());
 257         GraphEvent event = GraphEvent.builder(GraphEventOperation.UPDATE)
 258                 .vertex(GraphEventVertex.fromVertex(vertex, version)).build();
 259
 260         GraphEventEnvelope response = sendAndWait(event);
 261
 262         EntityTag entityTag;
 263         try {
 264             entityTag = new EntityTag(etagGenerator.computeHashForVertex(response.getBody().getVertex()));
 265         } catch (IOException e) {
 266             throw new CrudException(e);
 267         }
 268         String responsePayload = responseHandler.handleVertexResponse(version, event, response);
 269
 270         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 271     }
 272
 273     @Override
 274     public ImmutablePair<EntityTag, String> patchVertex(String version, String id, String type, VertexPayload payload)
 275             throws CrudException {
 276         OperationResult existingVertexOpResult = dao.getVertex(id, OxmModelValidator.resolveCollectionType(version, type), version,
 277                 new HashMap<String, String>());
 278         Vertex existingVertex = Vertex.fromJson(existingVertexOpResult.getResult(), version);
 279         Vertex patchedVertex = OxmModelValidator.validateIncomingPatchPayload(id, version, type,
 280                 payload.getProperties(), existingVertex);
 281         GraphEvent event = GraphEvent.builder(GraphEventOperation.UPDATE)
 282                 .vertex(GraphEventVertex.fromVertex(patchedVertex, version)).build();
 283
 284         GraphEventEnvelope response = sendAndWait(event);
 285
 286         EntityTag entityTag;
 287         try {
 288             entityTag = new EntityTag(etagGenerator.computeHashForVertex(response.getBody().getVertex()));
 289         } catch (IOException e) {
 290             throw new CrudException(e);
 291         }
 292         String responsePayload = responseHandler.handleVertexResponse(version, event, response);
 293
 294         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 295     }
 296
 297     @Override
 298     public String deleteVertex(String version, String id, String type) throws CrudException {
 299         type = OxmModelValidator.resolveCollectionType(version, type);
 300         GraphEvent event = GraphEvent.builder(GraphEventOperation.DELETE)
 301                 .vertex(new GraphEventVertex(id, version, type, null)).build();
 302
 303         GraphEventEnvelope response = sendAndWait(event);
 304         return responseHandler.handleDeletionResponse(event, response);
 305     }
 306
 307     @Override
 308     public String deleteEdge(String version, String id, String type) throws CrudException {
 309         RelationshipSchemaValidator.validateType(version, type);
 310         GraphEvent event = GraphEvent.builder(GraphEventOperation.DELETE)
 311                 .edge(new GraphEventEdge(id, version, type, null, null, null)).build();
 312
 313         GraphEventEnvelope response = sendAndWait(event);
 314         return responseHandler.handleDeletionResponse(event, response);
 315     }
 316
 317     @Override
 318     public ImmutablePair<EntityTag, String> updateEdge(String version, String id, String type, EdgePayload payload)
 319             throws CrudException {
 320         OperationResult operationResult = dao.getEdge(id, type, new HashMap<String, String>());
 321         Edge edge = Edge.fromJson(operationResult.getResult());
 322         Edge validatedEdge = RelationshipSchemaValidator.validateIncomingUpdatePayload(edge, version, payload);
 323         GraphEvent event = GraphEvent.builder(GraphEventOperation.UPDATE)
 324                 .edge(GraphEventEdge.fromEdge(validatedEdge, version)).build();
 325
 326         GraphEventEnvelope response = sendAndWait(event);
 327
 328         EntityTag entityTag;
 329         try {
 330             entityTag = new EntityTag(etagGenerator.computeHashForEdge(response.getBody().getEdge()));
 331         } catch (IOException e) {
 332             throw new CrudException(e);
 333         }
 334         String responsePayload = responseHandler.handleEdgeResponse(version, event, response);
 335
 336         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 337     }
 338
 339     @Override
 340     public ImmutablePair<EntityTag, String> patchEdge(String version, String id, String type, EdgePayload payload)
 341             throws CrudException {
 342         OperationResult operationResult = dao.getEdge(id, type, new HashMap<String, String>());
 343         Edge edge = Edge.fromJson(operationResult.getResult());
 344         Edge patchedEdge = RelationshipSchemaValidator.validateIncomingPatchPayload(edge, version, payload);
 345         GraphEvent event = GraphEvent.builder(GraphEventOperation.UPDATE)
 346                 .edge(GraphEventEdge.fromEdge(patchedEdge, version)).build();
 347
 348         GraphEventEnvelope response = sendAndWait(event);
 349
 350         EntityTag entityTag;
 351         try {
 352             entityTag = new EntityTag(etagGenerator.computeHashForEdge(response.getBody().getEdge()));
 353         } catch (IOException e) {
 354             throw new CrudException(e);
 355         }
 356         String responsePayload = responseHandler.handleEdgeResponse(version, event, response);
 357
 358         return new ImmutablePair<EntityTag, String>(entityTag, responsePayload);
 359     }
 360
 361     @PreDestroy
 362     protected void preShutdown() {
 363         timer.cancel();
 364     }
 365
 366     @Override
 367     protected Vertex addBulkVertex(Vertex vertex, String version, String dbTransId) throws CrudException {
 368         GraphEvent event = GraphEvent.builder(GraphEventOperation.CREATE)
 369                 .vertex(GraphEventVertex.fromVertex(vertex, version)).build();
 370         event.setDbTransactionId(dbTransId);
 371         GraphEvent response = publishEvent(event);
 372         return response.getVertex().toVertex();
 373     }
 374
 375     @Override
 376     protected Vertex updateBulkVertex(Vertex vertex, String id, String version, String dbTransId) throws CrudException {
 377         GraphEvent event = GraphEvent.builder(GraphEventOperation.UPDATE)
 378                 .vertex(GraphEventVertex.fromVertex(vertex, version)).build();
 379         event.setDbTransactionId(dbTransId);
 380         GraphEvent response = publishEvent(event);
 381         return response.getVertex().toVertex();
 382     }
 383
 384     @Override
 385     protected void deleteBulkVertex(String id, String version, String type, String dbTransId) throws CrudException {
 386         GraphEvent event = GraphEvent.builder(GraphEventOperation.DELETE)
 387                 .vertex(new GraphEventVertex(id, version, type, null)).build();
 388         event.setDbTransactionId(dbTransId);
 389         publishEvent(event);
 390     }
 391
 392     @Override
 393     protected Edge addBulkEdge(Edge edge, String version, String dbTransId) throws CrudException {
 394         GraphEvent event =
 395                 GraphEvent.builder(GraphEventOperation.CREATE).edge(GraphEventEdge.fromEdge(edge, version)).build();
 396         event.setDbTransactionId(dbTransId);
 397         GraphEvent response = publishEvent(event);
 398         return response.getEdge().toEdge();
 399     }
 400
 401     @Override
 402     protected Edge updateBulkEdge(Edge edge, String version, String dbTransId) throws CrudException {
 403         GraphEvent event =
 404                 GraphEvent.builder(GraphEventOperation.UPDATE).edge(GraphEventEdge.fromEdge(edge, version)).build();
 405         event.setDbTransactionId(dbTransId);
 406         GraphEvent response = publishEvent(event);
 407         return response.getEdge().toEdge();
 408     }
 409
 410     @Override
 411     protected void deleteBulkEdge(String id, String version, String type, String dbTransId) throws CrudException {
 412         GraphEvent event = GraphEvent.builder(GraphEventOperation.DELETE)
 413                 .edge(new GraphEventEdge(id, version, type, null, null, null)).build();
 414         event.setDbTransactionId(dbTransId);
 415         publishEvent(event);
 416     }
 417
 418     private GraphEvent publishEvent(GraphEvent event) throws CrudException {
 419         GraphEventEnvelope response = sendAndWait(event);
 420         responseHandler.handleBulkEventResponse(event, response);
 421         return response.getBody();
 422     }
 423 }