src/main/java/org/onap/aai/datarouter/policy/AbstractSpikeEntityEventProcessor.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.aai.datarouter.policy;
  22
  23 import java.io.FileNotFoundException;
  24 import java.io.IOException;
  25 import java.security.NoSuchAlgorithmException;
  26 import java.util.ArrayList;
  27 import java.util.Arrays;
  28 import java.util.Collection;
  29 import java.util.HashMap;
  30 import java.util.Iterator;
  31 import java.util.List;
  32 import java.util.Map;
  33
  34 import org.apache.camel.Exchange;
  35 import org.apache.camel.Processor;
  36 import org.eclipse.persistence.dynamic.DynamicType;
  37 import org.eclipse.persistence.internal.helper.DatabaseField;
  38 import org.eclipse.persistence.jaxb.dynamic.DynamicJAXBContext;
  39 import org.eclipse.persistence.oxm.MediaType;
  40 import org.json.JSONException;
  41 import org.json.JSONObject;
  42 import org.onap.aai.cl.api.Logger;
  43 import org.onap.aai.cl.eelf.LoggerFactory;
  44 import org.onap.aai.cl.mdc.MdcContext;
  45 import org.onap.aai.schema.OxmModelLoader;
  46 import org.onap.aai.datarouter.entity.DocumentStoreDataEntity;
  47 import org.onap.aai.datarouter.entity.SpikeEventEntity;
  48 import org.onap.aai.datarouter.entity.SpikeEventMeta;
  49 import org.onap.aai.datarouter.entity.SpikeEventVertex;
  50 import org.onap.aai.datarouter.logging.EntityEventPolicyMsgs;
  51 import org.onap.aai.datarouter.util.RouterServiceUtil;
  52 import org.onap.aai.datarouter.util.SearchServiceAgent;
  53 import org.onap.aai.entity.OxmEntityDescriptor;
  54 import org.onap.aai.util.EntityOxmReferenceHelper;
  55 import org.onap.aai.util.ExternalOxmModelProcessor;
  56 import org.onap.aai.util.Version;
  57 import org.onap.aai.util.VersionedOxmEntities;
  58 import org.onap.aai.restclient.client.Headers;
  59 import org.onap.aai.restclient.client.OperationResult;
  60 import org.onap.aai.restclient.rest.HttpUtil;
  61 import org.onap.aai.setup.SchemaVersions;
  62 import org.slf4j.MDC;
  63
  64 import com.fasterxml.jackson.core.JsonProcessingException;
  65 import com.fasterxml.jackson.databind.JsonNode;
  66 import com.fasterxml.jackson.databind.ObjectMapper;
  67
  68 public abstract class AbstractSpikeEntityEventProcessor implements Processor {
  69
  70   protected static final String additionalInfo = "Response of SpikeEntityEventPolicy";
  71   private Collection<ExternalOxmModelProcessor> externalOxmModelProcessors;
  72
  73   protected final String PROCESS_SPIKE_EVENT = "Process Spike Event";
  74
  75   protected static final String UPDATE_NOTIFICATION = "update-notification";
  76   protected static final String SPIKE = "SPIKE";
  77
  78   protected static final String HEADER_KEY = "header";
  79   protected static final String EVENT_TYPE_KEY = "event-type";
  80   protected static final String SOURCE_NAME_KEY = "source-name";
  81   protected static final String BODY_KEY = "body";
  82   protected static final String OPERATION_KEY = "operation";
  83
  84   protected static final String EDGE_KEY = "relationship";
  85
  86   protected static final String VERTEX_KEY = "vertex";
  87   protected static final String VERTEX_PROPERTIES_KEY = "properties";
  88   protected static final String VERTEX_KEY_KEY = "key";
  89   protected static final String VERTEX_TYPE_KEY = "type";
  90   protected static final String VERTEX_SCHEMA_VERSION_KEY = "schema-version";
  91
  92   protected static final String CREATE = "create";
  93   protected static final String DELETE = "delete";
  94   protected static final String UPDATE = "update";
  95
  96   protected enum ResponseType {
  97     SUCCESS, PARTIAL_SUCCESS, FAILURE;
  98   }
  99
 100   protected final List<String> SUPPORTED_ACTIONS = Arrays.asList(CREATE, UPDATE, DELETE);
 101
 102   protected Map<String, DynamicJAXBContext> oxmVersionContextMap = new HashMap<>();
 103   protected String oxmVersion = null;
 104
 105   /** Agent for communicating with the Search Service. */
 106   protected SearchServiceAgent searchAgent = null;
 107   protected String searchIndexName;
 108   protected String searchIndexSchema;
 109   protected String createIndexUrl;
 110
 111   protected Logger logger;
 112   protected Logger metricsLogger;
 113   protected ObjectMapper mapper;
 114
 115   public AbstractSpikeEntityEventProcessor(SpikeEventPolicyConfig config)
 116       throws FileNotFoundException {
 117     mapper = new ObjectMapper();
 118     LoggerFactory loggerFactoryInstance = LoggerFactory.getInstance();
 119     logger = loggerFactoryInstance.getLogger(AbstractSpikeEntityEventProcessor.class.getName());
 120     metricsLogger =
 121         loggerFactoryInstance.getMetricsLogger(AbstractSpikeEntityEventProcessor.class.getName());
 122
 123     // Instantiate the agent that we will use for interacting with the Search Service.
 124     searchAgent = new SearchServiceAgent(config.getSearchCertName(), config.getSearchKeystore(),
 125         config.getSearchKeystorePwd(), AbstractSpikeEntityEventProcessor
 126             .concatSubUri(config.getSearchBaseUrl(), config.getSearchEndpoint()),
 127         config.getSearchEndpointDocuments(), logger);
 128
 129     this.externalOxmModelProcessors = new ArrayList<>();
 130     this.externalOxmModelProcessors.add(EntityOxmReferenceHelper.getInstance());
 131     OxmModelLoader.registerExternalOxmModelProcessors(externalOxmModelProcessors);
 132     OxmModelLoader.loadModels(config.getSchemaVersions(), config.getSchemaLocationsBean());
 133     oxmVersionContextMap = OxmModelLoader.getVersionContextMap();
 134     parseLatestOxmVersion();
 135   }
 136
 137   public String getCreateIndexUrl() {
 138     return createIndexUrl;
 139   }
 140
 141
 142   public void setCreateIndexUrl(String createIndexUrl) {
 143     this.createIndexUrl = createIndexUrl;
 144   }
 145
 146   public String getSearchIndexName() {
 147     return searchIndexName;
 148   }
 149
 150
 151   public void setSearchIndexName(String searchIndexName) {
 152     this.searchIndexName = searchIndexName;
 153   }
 154
 155   public String getSearchIndexSchema() {
 156     return searchIndexSchema;
 157   }
 158
 159
 160   public void setSearchIndexSchema(String searchIndexSchema) {
 161     this.searchIndexSchema = searchIndexSchema;
 162   }
 163
 164   protected void startup() {
 165
 166   }
 167
 168   /*
 169    * Load the UEB JSON payload, any errors would result to a failure case response.
 170    */
 171   protected JSONObject getUebContentAsJson(String payload, String contentKey) {
 172
 173     JSONObject uebJsonObj;
 174     JSONObject uebObjContent;
 175
 176     try {
 177       uebJsonObj = new JSONObject(payload);
 178     } catch (JSONException e) {
 179       logger.debug(EntityEventPolicyMsgs.UEB_INVALID_PAYLOAD_JSON_FORMAT, payload);
 180       logger.error(EntityEventPolicyMsgs.UEB_INVALID_PAYLOAD_JSON_FORMAT, payload);
 181       return null;
 182     }
 183
 184     if (uebJsonObj.has(contentKey)) {
 185       uebObjContent = uebJsonObj.getJSONObject(contentKey);
 186     } else {
 187       logger.debug(EntityEventPolicyMsgs.UEB_FAILED_TO_PARSE_PAYLOAD, contentKey);
 188       logger.error(EntityEventPolicyMsgs.UEB_FAILED_TO_PARSE_PAYLOAD, contentKey);
 189       return null;
 190     }
 191
 192     return uebObjContent;
 193   }
 194   public abstract void process(Exchange exchange) throws Exception;
 195
 196
 197   private void parseLatestOxmVersion() {
 198     int latestVersion = -1;
 199     if (oxmVersionContextMap != null) {
 200       Iterator it = oxmVersionContextMap.entrySet().iterator();
 201       while (it.hasNext()) {
 202         Map.Entry pair = (Map.Entry) it.next();
 203
 204         String version = pair.getKey().toString();
 205         int versionNum = Integer.parseInt(version.substring(1, version.length()));
 206
 207         if (versionNum > latestVersion) {
 208           latestVersion = versionNum;
 209           oxmVersion = pair.getKey().toString();
 210         }
 211
 212         logger.info(EntityEventPolicyMsgs.PROCESS_OXM_MODEL_FOUND, pair.getKey().toString());
 213       }
 214     } else {
 215       logger.error(EntityEventPolicyMsgs.PROCESS_OXM_MODEL_MISSING, "");
 216     }
 217   }
 218
 219
 220
 221   /**
 222    * This will be used in: updateSearchEntityWithCrossEntityReference not this scope Convert object
 223    * to json.
 224    *
 225    * @param object the object
 226    * @param pretty the pretty
 227    * @return the string
 228    * @throws JsonProcessingException the json processing exception
 229    *
 230    *         protected static String convertObjectToJson(Object object, boolean pretty) throws
 231    *         JsonProcessingException { ObjectWriter ow;
 232    *
 233    *         if (pretty) { ow = new ObjectMapper().writer().withDefaultPrettyPrinter();
 234    *
 235    *         } else { ow = new ObjectMapper().writer(); }
 236    *
 237    *         return ow.writeValueAsString(object); }
 238    */
 239
 240   protected void returnWithError(Exchange exchange, String payload, String errorMsg) {
 241     logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE, errorMsg);
 242     logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, errorMsg, payload);
 243     setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 244   }
 245
 246   private boolean isJSONValid(String test) {
 247     try {
 248       new JSONObject(test);
 249     } catch (JSONException ex) {
 250       return false;
 251     }
 252     return true;
 253   }
 254
 255
 256
 257   protected String getSpikeEventAction(Exchange exchange, String uebPayload) {
 258     JSONObject mainJson = new JSONObject(uebPayload);
 259     String action = mainJson.getString(OPERATION_KEY);
 260     if (action == null || !SUPPORTED_ACTIONS.contains(action.toLowerCase())) {
 261       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 262           "Unrecognized action '" + action + "'", uebPayload);
 263       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 264           "Unrecognized action '" + action + "'");
 265       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 266       return null;
 267     }
 268     return action;
 269   }
 270
 271   protected String getExchangeBody(Exchange exchange) {
 272     String uebPayload = exchange.getIn().getBody().toString();
 273     if (uebPayload == null || !isJSONValid(uebPayload)) {
 274       uebPayload = exchange.getIn().getBody(String.class);
 275       if (uebPayload == null || !isJSONValid(uebPayload)) {
 276         returnWithError(exchange, uebPayload, "Invalid Payload");
 277         return null;
 278       }
 279     }
 280     return uebPayload;
 281   }
 282
 283   protected SpikeEventVertex populateEventVertex(Exchange exchange, String uebPayload)
 284       throws Exception {
 285
 286     // Load the UEB payload data, any errors will result in a failure and discard
 287
 288     JSONObject spikeObjVertex = getUebContentAsJson(uebPayload, VERTEX_KEY);
 289     if (spikeObjVertex == null) {
 290       returnWithError(exchange, uebPayload, "Payload is missing " + VERTEX_KEY);
 291       return null;
 292     }
 293
 294     SpikeEventVertex eventVertex = initializeSpikeEventVertex(spikeObjVertex.toString());
 295     return eventVertex;
 296   }
 297
 298   protected DynamicJAXBContext readOxm(Exchange exchange, String uebPayload) {
 299     DynamicJAXBContext oxmJaxbContext = loadOxmContext(oxmVersion);
 300     if (oxmJaxbContext == null) {
 301       logger.error(EntityEventPolicyMsgs.OXM_VERSION_NOT_SUPPORTED, oxmVersion);
 302       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "OXM version mismatch", uebPayload);
 303
 304       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 305       return null;
 306     }
 307     return oxmJaxbContext;
 308   }
 309
 310
 311   protected String getEntityType(Exchange exchange, SpikeEventVertex eventVertex,
 312       String uebPayload) {
 313
 314     String entityType = eventVertex.getType();
 315     if (entityType == null || entityType.isEmpty()) {
 316       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 317           "Payload header missing entity type", uebPayload);
 318       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 319           "Payload header missing entity type");
 320
 321       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 322       return null;
 323     }
 324     return entityType;
 325   }
 326
 327
 328
 329   protected String getEntityLink(Exchange exchange, SpikeEventVertex eventVertex,
 330       String uebPayload) {
 331     String entityKey = eventVertex.getKey();
 332     if (entityKey == null || entityKey.isEmpty()) {
 333       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "Payload vertex missing entity key",
 334           uebPayload);
 335       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 336           "Payload vertex missing entity key");
 337
 338       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 339       return null;
 340     }
 341     //EntityLink never can be null if  entityKey is not null. no need to check
 342     return eventVertex.getEntityLink();
 343
 344   }
 345
 346
 347
 348   /*
 349    * Use the OXM Model to determine the primary key field name based on the entity-type
 350    */
 351   protected SpikeEventEntity populateSpikeEventEntity(Exchange exchange,
 352       SpikeEventEntity spikeEventEntity, DynamicJAXBContext oxmJaxbContext, String entityType,
 353       String action, String uebPayload, String oxmEntityType, List<String> searchableAttr) {
 354
 355     String entityPrimaryKeyFieldName =
 356         getEntityPrimaryKeyFieldName(oxmJaxbContext, oxmEntityType, entityType);
 357     if (entityPrimaryKeyFieldName == null) {
 358       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 359           "Payload missing primary key attribute");
 360       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 361           "Payload missing primary key attribute", uebPayload);
 362       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 363       return null;
 364     }
 365     String entityPrimaryKeyFieldValue = lookupValueUsingKey(uebPayload, entityPrimaryKeyFieldName);
 366     if (entityPrimaryKeyFieldValue == null || entityPrimaryKeyFieldValue.isEmpty()) {
 367       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 368           "Payload missing primary value attribute");
 369       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 370           "Payload missing primary value attribute", uebPayload);
 371
 372       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 373       return null;
 374     }
 375
 376
 377     if (!getSearchTags(spikeEventEntity, searchableAttr, uebPayload, action)) {
 378       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 379           "Payload missing searchable attribute for entity type '" + entityType + "'");
 380       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 381           "Payload missing searchable attribute for entity type '" + entityType + "'", uebPayload);
 382
 383       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 384       return null;
 385     }
 386     spikeEventEntity.setEntityPrimaryKeyName(entityPrimaryKeyFieldName);
 387     spikeEventEntity.setEntityPrimaryKeyValue(entityPrimaryKeyFieldName);
 388
 389     try {
 390       spikeEventEntity.deriveFields();
 391
 392     } catch (NoSuchAlgorithmException e) {
 393       logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "Cannot create unique SHA digest");
 394       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "Cannot create unique SHA digest",
 395           uebPayload);
 396
 397       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 398       return null;
 399     }
 400     return spikeEventEntity;
 401   }
 402
 403   protected void setResponse(Exchange exchange, ResponseType responseType, String additionalInfo) {
 404
 405     exchange.getOut().setHeader("ResponseType", responseType.toString());
 406     exchange.getOut().setBody(additionalInfo);
 407   }
 408
 409
 410  protected String getOxmEntityType(String entityType) {
 411    return new OxmEntityTypeConverter().convert(entityType);
 412  }
 413
 414  protected List<String> getSearchableAttibutes(DynamicJAXBContext oxmJaxbContext, String oxmEntityType,
 415      String entityType, String uebPayload,Exchange exchange) {
 416    List<String> searchableAttr =
 417        getOxmAttributes(oxmJaxbContext, oxmEntityType, entityType, "searchable");
 418    if (searchableAttr == null) {
 419      logger.error(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 420          "Searchable attribute not found for payload entity type '" + entityType + "'");
 421      logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE,
 422          "Searchable attribute not found for payload entity type '" + entityType + "'",
 423          uebPayload);
 424
 425      setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 426      return null;
 427    }
 428    return searchableAttr;
 429  }
 430
 431
 432   private SpikeEventVertex initializeSpikeEventVertex(String payload) {
 433
 434     SpikeEventVertex eventVertex = null;
 435     ObjectMapper mapper = new ObjectMapper();
 436
 437     // Make sure that were were actually passed in a valid string.
 438     if (payload == null || payload.isEmpty()) {
 439       logger.debug(EntityEventPolicyMsgs.UEB_FAILED_TO_PARSE_PAYLOAD, VERTEX_KEY);
 440       logger.error(EntityEventPolicyMsgs.UEB_FAILED_TO_PARSE_PAYLOAD, VERTEX_KEY);
 441
 442       return eventVertex;
 443     }
 444
 445     // Marshal the supplied string into a UebEventHeader object.
 446     try {
 447       eventVertex = mapper.readValue(payload, SpikeEventVertex.class);
 448     } catch (JsonProcessingException e) {
 449       logger.error(EntityEventPolicyMsgs.UEB_FAILED_UEBEVENTHEADER_CONVERSION, e.toString());
 450     } catch (Exception e) {
 451       logger.error(EntityEventPolicyMsgs.UEB_FAILED_UEBEVENTHEADER_CONVERSION, e.toString());
 452     }
 453
 454     if (eventVertex != null) {
 455       logger.debug(EntityEventPolicyMsgs.UEB_EVENT_HEADER_PARSED, eventVertex.toString());
 456     }
 457
 458     return eventVertex;
 459   }
 460
 461
 462   private String getEntityPrimaryKeyFieldName(DynamicJAXBContext oxmJaxbContext,
 463       String oxmEntityType, String entityType) {
 464
 465     DynamicType entity = oxmJaxbContext.getDynamicType(oxmEntityType);
 466     if (entity == null) {
 467       return null;
 468     }
 469
 470     List<DatabaseField> list = entity.getDescriptor().getPrimaryKeyFields();
 471     if (list != null && !list.isEmpty()) {
 472       String keyName = list.get(0).getName();
 473       return keyName.substring(0, keyName.indexOf('/'));
 474     }
 475
 476     return "";
 477   }
 478
 479   private String lookupValueUsingKey(String payload, String key) throws JSONException {
 480     JsonNode jsonNode = convertToJsonNode(payload);
 481     return RouterServiceUtil.recursivelyLookupJsonPayload(jsonNode, key);
 482   }
 483
 484
 485   private JsonNode convertToJsonNode(String payload) {
 486
 487     ObjectMapper mapper = new ObjectMapper();
 488     JsonNode jsonNode = null;
 489     try {
 490       jsonNode = mapper.readTree(mapper.getJsonFactory().createJsonParser(payload));
 491     } catch (IOException e) {
 492       logger.debug(EntityEventPolicyMsgs.FAILED_TO_PARSE_UEB_PAYLOAD, VERTEX_KEY + " missing",
 493           payload);
 494       logger.error(EntityEventPolicyMsgs.FAILED_TO_PARSE_UEB_PAYLOAD, VERTEX_KEY + " missing",
 495           "");
 496     }
 497
 498     return jsonNode;
 499   }
 500
 501
 502   private boolean getSearchTags(SpikeEventEntity spikeEventEntity, List<String> searchableAttr,
 503       String payload, String action) {
 504
 505     boolean hasSearchableAttr = false;
 506     for (String searchTagField : searchableAttr) {
 507       String searchTagValue;
 508       if (searchTagField.equalsIgnoreCase(spikeEventEntity.getEntityPrimaryKeyName())) {
 509         searchTagValue = spikeEventEntity.getEntityPrimaryKeyValue();
 510       } else {
 511         searchTagValue = this.lookupValueUsingKey(payload, searchTagField);
 512       }
 513
 514       if (searchTagValue != null && !searchTagValue.isEmpty()) {
 515         hasSearchableAttr = true;
 516         spikeEventEntity.addSearchTagWithKey(searchTagValue, searchTagField);
 517       }
 518     }
 519     return hasSearchableAttr;
 520   }
 521
 522   /*
 523    * Check if OXM version is available. If available, load it.
 524    */
 525   private DynamicJAXBContext loadOxmContext(String version) {
 526     if (version == null) {
 527       logger.error(EntityEventPolicyMsgs.FAILED_TO_FIND_OXM_VERSION, version);
 528       return null;
 529     }
 530
 531     return oxmVersionContextMap.get(version);
 532   }
 533
 534   private List<String> getOxmAttributes(DynamicJAXBContext oxmJaxbContext, String oxmEntityType,
 535       String entityType, String fieldName) {
 536
 537     DynamicType entity = oxmJaxbContext.getDynamicType(oxmEntityType);
 538     if (entity == null) {
 539       return null;
 540     }
 541
 542     /*
 543      * Check for searchable XML tag
 544      */
 545     List<String> fieldValues = null;
 546     Map<String, String> properties = entity.getDescriptor().getProperties();
 547     for (Map.Entry<String, String> entry : properties.entrySet()) {
 548       if (entry.getKey().equalsIgnoreCase(fieldName)) {
 549         fieldValues = Arrays.asList(entry.getValue().split(","));
 550         break;
 551       }
 552     }
 553
 554     return fieldValues;
 555   }
 556
 557   protected SpikeEventEntity getPopulatedEntity(JsonNode entityNode,
 558       OxmEntityDescriptor resultDescriptor) {
 559     SpikeEventEntity d = new SpikeEventEntity();
 560
 561     d.setEntityType(resultDescriptor.getEntityName());
 562
 563     List<String> primaryKeyValues = new ArrayList<>();
 564     List<String> primaryKeyNames = new ArrayList<>();
 565     String pkeyValue;
 566
 567     for (String keyName : resultDescriptor.getPrimaryKeyAttributeName()) {
 568       pkeyValue = RouterServiceUtil.getNodeFieldAsText(entityNode, keyName);
 569       if (pkeyValue != null) {
 570         primaryKeyValues.add(pkeyValue);
 571         primaryKeyNames.add(keyName);
 572       } else {
 573         // logger.warn("getPopulatedDocument(), pKeyValue is null for entityType = " +
 574         // resultDescriptor.getEntityName());
 575         logger.error(EntityEventPolicyMsgs.PRIMARY_KEY_NULL_FOR_ENTITY_TYPE,
 576             resultDescriptor.getEntityName());
 577       }
 578     }
 579
 580     final String primaryCompositeKeyValue = RouterServiceUtil.concatArray(primaryKeyValues, "/");
 581     d.setEntityPrimaryKeyValue(primaryCompositeKeyValue);
 582     final String primaryCompositeKeyName = RouterServiceUtil.concatArray(primaryKeyNames, "/");
 583     d.setEntityPrimaryKeyName(primaryCompositeKeyName);
 584
 585     final List<String> searchTagFields = resultDescriptor.getSearchableAttributes();
 586
 587     /*
 588      * Based on configuration, use the configured field names for this entity-Type to build a
 589      * multi-value collection of search tags for elastic search entity search criteria.
 590      */
 591
 592
 593     for (String searchTagField : searchTagFields) {
 594       String searchTagValue = RouterServiceUtil.getNodeFieldAsText(entityNode, searchTagField);
 595       if (searchTagValue != null && !searchTagValue.isEmpty()) {
 596         d.addSearchTagWithKey(searchTagValue, searchTagField);
 597       }
 598     }
 599
 600     return d;
 601   }
 602
 603
 604
 605   // put this here until we find a better spot
 606   /**
 607    * Helper utility to concatenate substrings of a URI together to form a proper URI.
 608    *
 609    * @param suburis the list of substrings to concatenate together
 610    * @return the concatenated list of substrings
 611    */
 612   private static String concatSubUri(String... suburis) {
 613     String finalUri = "";
 614
 615     for (String suburi : suburis) {
 616
 617       if (suburi != null) {
 618         // Remove any leading / since we only want to append /
 619         suburi = suburi.replaceFirst("^/*", "");
 620
 621         // Add a trailing / if one isn't already there
 622         finalUri += suburi.endsWith("/") ? suburi : suburi + "/";
 623       }
 624     }
 625
 626     return finalUri;
 627   }
 628
 629
 630
 631   /**
 632    * Perform create, read, update or delete (CRUD) operation on search engine's suggestive search
 633    * index
 634    *
 635    * @param eventEntity Entity/data to use in operation
 636    * @param action The operation to perform
 637    */
 638   protected void handleSearchServiceOperation(DocumentStoreDataEntity eventEntity, String action,
 639       String index) {
 640     try {
 641
 642       Map<String, List<String>> headers = new HashMap<>();
 643       headers.put(Headers.FROM_APP_ID, Arrays.asList("DataLayer"));
 644       headers.put(Headers.TRANSACTION_ID, Arrays.asList(MDC.get(MdcContext.MDC_REQUEST_ID)));
 645
 646       String entityId = eventEntity.getId();
 647
 648       if ((action.equalsIgnoreCase(CREATE) && entityId != null)
 649           || action.equalsIgnoreCase(UPDATE)) {
 650
 651         // Run the GET to retrieve the ETAG from the search service
 652         OperationResult storedEntity = searchAgent.getDocument(index, entityId);
 653
 654         if (HttpUtil.isHttpResponseClassSuccess(storedEntity.getResultCode())) {
 655           List<String> etag = storedEntity.getHeaders().get(Headers.ETAG);
 656
 657           if (etag != null && !etag.isEmpty()) {
 658             headers.put(Headers.IF_MATCH, etag);
 659           } else {
 660             logger.error(EntityEventPolicyMsgs.NO_ETAG_AVAILABLE_FAILURE, index, entityId);
 661           }
 662         }
 663
 664         // Write the entity to the search service.
 665         // PUT
 666         searchAgent.putDocument(index, entityId, eventEntity.getAsJson(), headers);
 667       } else if (action.equalsIgnoreCase(CREATE)) {
 668         // Write the entry to the search service.
 669         searchAgent.postDocument(index, eventEntity.getAsJson(), headers);
 670
 671       } else if (action.equalsIgnoreCase(DELETE)) {
 672         // Run the GET to retrieve the ETAG from the search service
 673         OperationResult storedEntity = searchAgent.getDocument(index, entityId);
 674
 675         if (HttpUtil.isHttpResponseClassSuccess(storedEntity.getResultCode())) {
 676           List<String> etag = storedEntity.getHeaders().get(Headers.ETAG);
 677
 678           if (etag != null && !etag.isEmpty()) {
 679             headers.put(Headers.IF_MATCH, etag);
 680           } else {
 681             logger.error(EntityEventPolicyMsgs.NO_ETAG_AVAILABLE_FAILURE, index, entityId);
 682           }
 683
 684           /*
 685            * The Spring-Boot version of the search-data-service rejects the DELETE operation unless
 686            * we specify a Content-Type.
 687            */
 688
 689           headers.put("Content-Type", Arrays.asList(MediaType.APPLICATION_JSON.getMediaType()));
 690
 691           searchAgent.deleteDocument(index, eventEntity.getId(), headers);
 692         } else {
 693           logger.error(EntityEventPolicyMsgs.NO_ETAG_AVAILABLE_FAILURE, index, entityId);
 694         }
 695       } else {
 696         logger.error(EntityEventPolicyMsgs.ENTITY_OPERATION_NOT_SUPPORTED, action);
 697       }
 698     } catch (IOException e) {
 699       logger.error(EntityEventPolicyMsgs.FAILED_TO_UPDATE_ENTITY_IN_DOCSTORE, eventEntity.getId(),
 700           action);
 701     }
 702   }
 703
 704   protected SpikeEventMeta processSpikeEvent(Exchange exchange) {
 705
 706     SpikeEventMeta meta = new SpikeEventMeta();
 707     Object eventPayloadObj = null;
 708     String eventPayload = null;
 709     try {
 710        eventPayloadObj = exchange.getIn().getBody();
 711
 712       /*
 713        * It is expected that mainJson will have multiple top level objects: - header - body - result
 714        */
 715       if (eventPayloadObj == null) {
 716         returnWithError(exchange, null, "Invalid Payload");
 717         return null;
 718       }
 719
 720       eventPayload = (String)eventPayloadObj;
 721
 722       meta.setEventEntity(new JSONObject(eventPayload));
 723     } catch (JSONException exc) {
 724       returnWithError(exchange, eventPayload, "Invalid Payload");
 725       return null;
 726     }
 727
 728     JSONObject eventHeader = meta.getEventEntity().getJSONObject(HEADER_KEY);
 729
 730     if (eventHeader == null) {
 731       returnWithError(exchange, eventPayload, "Payload is missing " + HEADER_KEY);
 732       return null;
 733     }
 734
 735     meta.setEventHeader(eventHeader);
 736
 737     /*
 738      * Only process SPIKE update-notification events
 739      */
 740
 741     final String sourceName = eventHeader.getString(SOURCE_NAME_KEY);
 742     final String eventType = eventHeader.getString(EVENT_TYPE_KEY);
 743
 744     if (!(SPIKE.equals(sourceName) && UPDATE_NOTIFICATION.equals(eventType))) {
 745       // drop event
 746       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "Ignoring event with sourceName='"
 747           + sourceName + "' and eventType='" + eventType + "'.  Payload=" + eventPayload);
 748       logger.info(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 749           "Ignoring event with sourceName='" + sourceName + "' and eventType='" + eventType + "'.");
 750
 751       /*
 752        * I don't think ignoring a non-applicable event constitutes a failure.
 753        */
 754
 755       setResponse(exchange, ResponseType.SUCCESS, additionalInfo);
 756       return null;
 757     }
 758
 759     JSONObject eventBody = meta.getEventEntity().getJSONObject(BODY_KEY);
 760
 761     if (eventBody == null) {
 762       returnWithError(exchange, eventPayload, "Payload is missing " + BODY_KEY);
 763       return null;
 764     }
 765
 766     meta.setEventBody(eventBody);
 767
 768     String action = eventBody.getString(OPERATION_KEY);
 769     if (action == null || !SUPPORTED_ACTIONS.contains(action.toLowerCase())) {
 770       returnWithError(exchange, eventPayload, "Unrecognized action '" + action + "'");
 771       return null;
 772     }
 773     meta.setBodyOperationType(action);
 774
 775
 776     /*
 777      * Ignore spike events for edges
 778      */
 779     if (eventBody.has(EDGE_KEY)){
 780       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "Ignoring event for edge with sourceName='"
 781           + sourceName + "' and eventType='" + eventType + "'.  Payload=" + eventPayload);
 782       logger.info(EntityEventPolicyMsgs.DISCARD_EVENT_NONVERBOSE,
 783           "Ignoring event for edge with sourceName='" + sourceName + "' and eventType='" + eventType + "'.");
 784       setResponse(exchange, ResponseType.SUCCESS, additionalInfo);
 785       return null;
 786     }
 787
 788
 789     // Load the event body data, any errors will result in a failure and discard
 790     JSONObject spikeVertex = null;
 791     try {
 792       spikeVertex = eventBody.getJSONObject(VERTEX_KEY);
 793     } catch (JSONException exc) {
 794       returnWithError(exchange, eventPayload, "Payload is missing " + VERTEX_KEY);
 795       return null;
 796     }
 797
 798     meta.setSpikeVertex(spikeVertex);
 799
 800     SpikeEventVertex spikeEventVertex = null;
 801     try {
 802       spikeEventVertex = initializeSpikeEventVertex(spikeVertex);
 803     } catch (JSONException exc) {
 804       returnWithError(exchange, eventPayload, "Error initializating spike event.  Error: " + exc.getMessage());
 805       return null;
 806     }
 807
 808     meta.setSpikeEventVertex(spikeEventVertex);
 809
 810     DynamicJAXBContext oxmJaxbContext = loadOxmContext(oxmVersion);
 811     if (oxmJaxbContext == null) {
 812       logger.error(EntityEventPolicyMsgs.OXM_VERSION_NOT_SUPPORTED, oxmVersion);
 813       logger.debug(EntityEventPolicyMsgs.DISCARD_EVENT_VERBOSE, "OXM version mismatch",
 814           eventPayload);
 815
 816       setResponse(exchange, ResponseType.FAILURE, additionalInfo);
 817       return null;
 818     }
 819
 820     meta.setOxmJaxbContext(oxmJaxbContext);
 821
 822     String entityType = spikeEventVertex.getType();
 823     if (entityType == null || entityType.isEmpty()) {
 824       returnWithError(exchange, eventPayload, "Payload vertex missing entity type");
 825       return null;
 826     }
 827
 828     /*
 829      * test if entityType is in the model
 830      */
 831
 832     VersionedOxmEntities oxmEntities =
 833     EntityOxmReferenceHelper.getInstance().getVersionedOxmEntities(Version.valueOf(oxmVersion.toLowerCase()));
 834
 835     if (oxmEntities != null && !oxmEntities.getEntityTypeLookup().containsKey(entityType)) {
 836       returnWithError(exchange, eventPayload, "No matching OXM Descriptor for entity-type='" + entityType + "'");
 837       return null;
 838     }
 839
 840
 841     String entityKey = spikeEventVertex.getKey();
 842     if (entityKey == null || entityKey.isEmpty()) {
 843       returnWithError(exchange, eventPayload, "Payload vertex missing entity key");
 844       return null;
 845     }
 846
 847     String entityLink = spikeEventVertex.getEntityLink();
 848     if (entityLink == null || entityLink.isEmpty()) {
 849       returnWithError(exchange, eventPayload, "Payload vertex missing entity link");
 850       return null;
 851     }
 852
 853     JSONObject vertexProperties = null;
 854     try {
 855       vertexProperties = spikeVertex.getJSONObject(VERTEX_PROPERTIES_KEY);
 856     } catch (JSONException exc) {
 857       returnWithError(exchange, eventPayload, "Payload vertex missing " + VERTEX_PROPERTIES_KEY);
 858       return null;
 859     }
 860
 861     meta.setVertexProperties(vertexProperties);
 862
 863     // log the fact that all data are in good shape
 864     logger.info(EntityEventPolicyMsgs.PROCESS_ENTITY_EVENT_POLICY_NONVERBOSE, action, entityType);
 865     logger.debug(EntityEventPolicyMsgs.PROCESS_ENTITY_EVENT_POLICY_VERBOSE, action, entityType,
 866         eventPayload);
 867
 868     return meta;
 869
 870   }
 871
 872   protected SpikeEventVertex initializeSpikeEventVertex(JSONObject vertexObject) throws JSONException {
 873
 874     /*
 875      * These are all critical keys
 876      */
 877
 878     final String vertexType = vertexObject.getString(VERTEX_TYPE_KEY);
 879     final String vertexKey = vertexObject.getString(VERTEX_KEY_KEY);
 880     final String vertexSchemaVersion = vertexObject.getString(VERTEX_SCHEMA_VERSION_KEY);
 881
 882     SpikeEventVertex eventVertex = new SpikeEventVertex(vertexType, vertexKey);
 883     eventVertex.setSchemaVersion(vertexSchemaVersion);
 884     logger.debug(EntityEventPolicyMsgs.UEB_EVENT_HEADER_PARSED, eventVertex.toString());
 885
 886     return eventVertex;
 887
 888   }
 889
 890 }