ONAP-PAP-REST/src/main/java/org/onap/policy/pap/xacml/rest/elk/client/ElkConnectorImpl.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP-PAP-REST
   4  * ================================================================================
   5  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20 package org.onap.policy.pap.xacml.rest.elk.client;
  21
  22 import java.io.IOException;
  23 import java.util.Map;
  24 import java.util.Map.Entry;
  25
  26 import org.elasticsearch.index.query.QueryBuilders;
  27 import org.elasticsearch.index.query.QueryStringQueryBuilder;
  28 import org.elasticsearch.search.builder.SearchSourceBuilder;
  29 import org.json.JSONObject;
  30 import org.onap.policy.common.logging.flexlogger.FlexLogger;
  31 import org.onap.policy.common.logging.flexlogger.Logger;
  32 import org.onap.policy.rest.adapter.PolicyRestAdapter;
  33 import org.onap.policy.xacml.api.XACMLErrorConstants;
  34
  35 import io.searchbox.action.Action;
  36 import io.searchbox.client.JestClient;
  37 import io.searchbox.client.JestClientFactory;
  38 import io.searchbox.client.JestResult;
  39 import io.searchbox.client.config.HttpClientConfig;
  40 import io.searchbox.core.Delete;
  41 import io.searchbox.core.Index;
  42 import io.searchbox.core.Search;
  43 import io.searchbox.core.Search.Builder;
  44 import io.searchbox.indices.IndicesExists;
  45 import io.searchbox.indices.type.TypeExist;
  46 import io.searchbox.params.Parameters;
  47
  48 public class ElkConnectorImpl implements ElkConnector{
  49
  50     private static final Logger LOGGER = FlexLogger.getLogger(ElkConnector.class);
  51
  52     protected final JestClientFactory jestFactory = new JestClientFactory();
  53     protected final JestClient jestClient;
  54     protected static int QUERY_MAXRECORDS = 1000;
  55
  56     public ElkConnectorImpl() {
  57         if (LOGGER.isDebugEnabled()){
  58             LOGGER.debug("ENTER: -");
  59         }
  60         HttpClientConfig jestClientConfig = new HttpClientConfig.Builder(ELK_URL).multiThreaded(true).build();
  61         jestFactory.setHttpClientConfig(jestClientConfig);
  62         jestClient = jestFactory.getObject();
  63     }
  64
  65     protected boolean isType(PolicyIndexType type) throws IOException {
  66         if (LOGGER.isDebugEnabled()){
  67             LOGGER.debug("ENTER: -");
  68         }
  69
  70         try {
  71             Action<JestResult> typeQuery = new TypeExist.Builder(ELK_INDEX_POLICY).addType(type.toString()).build();
  72             JestResult result = jestClient.execute(typeQuery);
  73
  74             if (LOGGER.isInfoEnabled()) {
  75                 LOGGER.info("JSON:" + result.getJsonString());
  76                 LOGGER.info("ERROR:" + result.getErrorMessage());
  77                 LOGGER.info("PATH:" + result.getPathToResult());
  78                 LOGGER.info(result.getJsonObject());
  79             }
  80             return result.isSucceeded();
  81         } catch (IOException e) {
  82             LOGGER.warn("Error checking type existance of " + type.toString() + ": " + e.getMessage(), e);
  83             throw e;
  84         }
  85     }
  86
  87     protected boolean isIndex() throws IOException {
  88         try {
  89             Action<JestResult> indexQuery = new IndicesExists.Builder(ELK_INDEX_POLICY).build();
  90
  91             JestResult result = jestClient.execute(indexQuery);
  92             if (LOGGER.isInfoEnabled()) {
  93                 LOGGER.info("JSON:" + result.getJsonString());
  94                 LOGGER.info("ERROR:" + result.getErrorMessage());
  95                 LOGGER.info("PATH:" + result.getPathToResult());
  96                 LOGGER.info(result.getJsonObject());
  97             }
  98             return result.isSucceeded();
  99         } catch (IOException e) {
 100             LOGGER.warn("Error checking index existance of " + ELK_INDEX_POLICY + ": " + e.getMessage(), e);
 101             throw e;
 102         }
 103     }
 104     private boolean isAlphaNumeric(String query){
 105         return query.matches("[a-zA-Z_0-9]+");
 106     }
 107
 108     @Override
 109     public JestResult search(PolicyIndexType type, String text) throws IllegalStateException, IllegalArgumentException {
 110         if (LOGGER.isTraceEnabled()){
 111             LOGGER.trace("ENTER: " + text);
 112         }
 113
 114         if (text == null || text.isEmpty()) {
 115             throw new IllegalArgumentException("No search string provided");
 116         }
 117
 118         if(!isAlphaNumeric(text)){
 119             throw new IllegalArgumentException("Search must be alpha numeric");
 120         }
 121
 122         QueryStringQueryBuilder mQ = QueryBuilders.queryStringQuery("*"+text+"*");
 123         SearchSourceBuilder searchSourceBuilder =
 124                 new SearchSourceBuilder().query(mQ);
 125
 126         Builder searchBuilder = new Search.Builder(searchSourceBuilder.toString()).
 127                 addIndex(ELK_INDEX_POLICY).
 128                 setParameter(Parameters.SIZE, ElkConnectorImpl.QUERY_MAXRECORDS);
 129
 130         if (type == null || type == PolicyIndexType.all) {
 131             for (PolicyIndexType pT: PolicyIndexType.values()) {
 132                 if (pT != PolicyIndexType.all) {
 133                     searchBuilder.addType(pT.toString());
 134                 }
 135             }
 136         } else {
 137             searchBuilder.addType(type.toString());
 138         }
 139
 140         Search search = searchBuilder.build();
 141         JestResult result;
 142         try {
 143             result = jestClient.execute(search);
 144         } catch (IOException ioe) {
 145             LOGGER.warn(XACMLErrorConstants.ERROR_SYSTEM_ERROR + ":" +
 146                     search + ": " + ioe.getMessage(), ioe);
 147             throw new IllegalStateException(ioe);
 148         }
 149
 150         if (result.isSucceeded()) {
 151             if (LOGGER.isInfoEnabled()){
 152                 LOGGER.info("OK:" + result.getResponseCode() + ":" + search + ": " +
 153                         result.getPathToResult() + ":" + System.lineSeparator() +
 154                         result.getJsonString());
 155             }
 156         } else {
 157             /* Unsuccessful search */
 158             if (LOGGER.isWarnEnabled()){
 159                 LOGGER.warn(XACMLErrorConstants.ERROR_PROCESS_FLOW + ":" +
 160                         result.getResponseCode() + ": " +
 161                         search.getURI() + ":" +
 162                         result.getPathToResult() + ":" +
 163                         result.getJsonString() + ":" +
 164                         result.getErrorMessage());
 165             }
 166
 167             String errorMessage = result.getErrorMessage();
 168             if (errorMessage != null && !errorMessage.isEmpty()) {
 169                 String xMessage;
 170                 if (errorMessage.contains("TokenMgrError")) {
 171                     int indexError = errorMessage.lastIndexOf("TokenMgrError");
 172                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 173                 } else if (errorMessage.contains("QueryParsingException")) {
 174                     int indexError = errorMessage.lastIndexOf("QueryParsingException");
 175                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 176                 } else if (errorMessage.contains("JsonParseException")) {
 177                     int indexError = errorMessage.lastIndexOf("JsonParseException");
 178                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 179                 } else if (errorMessage.contains("Parse Failure")) {
 180                     int indexError = errorMessage.lastIndexOf("Parse Failure");
 181                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 182                 } else if (errorMessage.contains("SearchParseException")) {
 183                     int indexError = errorMessage.lastIndexOf("SearchParseException");
 184                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 185                 } else {
 186                     xMessage = result.getErrorMessage();
 187                 }
 188                 throw new IllegalStateException(xMessage);
 189             }
 190         }
 191
 192         return result;
 193     }
 194
 195
 196     @Override
 197     public JestResult search(PolicyIndexType type, String text,
 198             Map<String, String> filter_s)
 199                     throws IllegalStateException, IllegalArgumentException {
 200         if (LOGGER.isTraceEnabled()){
 201             LOGGER.trace("ENTER: " + text);
 202         }
 203
 204         if (filter_s == null || filter_s.size() == 0) {
 205             return search(type, text);
 206         }
 207
 208         if(!isAlphaNumeric(text)){
 209             throw new IllegalArgumentException("Search must be alpha numeric");
 210         }
 211
 212         String matches_s = "";
 213         matches_s = "{\n" +
 214                 "    \"size\" : "+ ElkConnectorImpl.QUERY_MAXRECORDS + ",\n" +
 215                 "    \"query\": {\n" +
 216                 "        \"bool\" : {\n" +
 217                 "            \"must\" : [";
 218
 219         String match_params = "";
 220         boolean first = true;
 221         for(Entry<String, String> entry : filter_s.entrySet()){
 222             String key = entry.getKey();
 223             String value = entry.getValue();
 224             if(first){
 225                 match_params = "\"match\" : {\""+key+"\" : \""+value+"\" }},";
 226                 first = false;
 227             }else{
 228                 match_params = match_params + "{\"match\" : { \""+key+"\" : \""+value+"\" } },";
 229             }
 230         }
 231         if(match_params.endsWith(",")){
 232             match_params = match_params.substring(0, match_params.length()-2);
 233         }
 234
 235         matches_s = matches_s + "{\n" + match_params + "\n}" ;
 236
 237         boolean query = false;
 238         String query_String = "";
 239         if(text != null){
 240             query = true;
 241             query_String = "{\n \"query_string\" : {\n \"query\" : \"*"+text+"*\"\n} \n}";
 242         }
 243
 244         if(query){
 245             matches_s = matches_s + "," +  query_String + "]\n}\n}\n}";
 246         }else{
 247             matches_s = matches_s + "]\n}\n}\n}";
 248         }
 249
 250         Builder searchBuilder = new Search.Builder(matches_s).addIndex(ELK_INDEX_POLICY);
 251
 252         if (type == null || type == PolicyIndexType.all) {
 253             for (PolicyIndexType pT: PolicyIndexType.values()) {
 254                 if (pT != PolicyIndexType.all) {
 255                     searchBuilder.addType(pT.toString());
 256                 }
 257             }
 258         } else {
 259             searchBuilder.addType(type.toString());
 260         }
 261
 262         Search search = searchBuilder.build();
 263
 264         JestResult result;
 265         try {
 266             result = jestClient.execute(search);
 267         } catch (IOException ioe) {
 268             LOGGER.warn(XACMLErrorConstants.ERROR_SYSTEM_ERROR + ":" +
 269                     search + ": " + ioe.getMessage(), ioe);
 270             throw new IllegalStateException(ioe);
 271         }
 272
 273         if (result.isSucceeded()) {
 274             if (LOGGER.isInfoEnabled()){
 275                 LOGGER.info("OK:" + result.getResponseCode() + ":" + search + ": " +
 276                         result.getPathToResult() + ":" + System.lineSeparator() +
 277                         result.getJsonString());
 278             }
 279         } else {
 280             /* Unsuccessful search */
 281             if (LOGGER.isWarnEnabled()){
 282                 LOGGER.warn(XACMLErrorConstants.ERROR_PROCESS_FLOW + ":" +
 283                         result.getResponseCode() + ": " +
 284                         search.getURI() + ":" +
 285                         result.getPathToResult() + ":" +
 286                         result.getJsonString() + ":" +
 287                         result.getErrorMessage());
 288             }
 289
 290             String errorMessage = result.getErrorMessage();
 291             if (errorMessage != null && !errorMessage.isEmpty()) {
 292                 String xMessage = errorMessage;
 293                 if (errorMessage.contains("TokenMgrError")) {
 294                     int indexError = errorMessage.lastIndexOf("TokenMgrError");
 295                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 296                 } else if (errorMessage.contains("QueryParsingException")) {
 297                     int indexError = errorMessage.lastIndexOf("QueryParsingException");
 298                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 299                 } else if (errorMessage.contains("JsonParseException")) {
 300                     int indexError = errorMessage.lastIndexOf("JsonParseException");
 301                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 302                 } else if (errorMessage.contains("Parse Failure")) {
 303                     int indexError = errorMessage.lastIndexOf("Parse Failure");
 304                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 305                 } else if (errorMessage.contains("SearchParseException")) {
 306                     int indexError = errorMessage.lastIndexOf("SearchParseException");
 307                     xMessage = "Invalid Search Expression.  Details: " + errorMessage.substring(indexError);
 308                 } else {
 309                     xMessage = result.getErrorMessage();
 310                 }
 311                 throw new IllegalStateException(xMessage);
 312             }
 313         }
 314
 315         return result;
 316     }
 317
 318     public boolean put(PolicyRestAdapter policyData)
 319             throws IOException, IllegalStateException {
 320         if (LOGGER.isTraceEnabled()) LOGGER.trace("ENTER");
 321
 322         PolicyIndexType indexType;
 323         try {
 324             String policyName = policyData.getNewFileName();
 325             if(policyName.contains("Config_")){
 326                 policyName = policyName.replace(".Config_", ":Config_");
 327             }else if(policyName.contains("Action_")){
 328                 policyName = policyName.replace(".Action_", ":Action_");
 329             }else if(policyName.contains("Decision_")){
 330                 policyName = policyName.replace(".Decision_", ":Decision_");
 331             }
 332
 333             String[] splitPolicyName = policyName.split(":");
 334             indexType = ElkConnector.toPolicyIndexType(splitPolicyName[1]);
 335         } catch (IllegalArgumentException e) {
 336             LOGGER.error(e);
 337             throw new IllegalStateException("ELK: Index: " + ELK_INDEX_POLICY + e.getMessage());
 338         }
 339         PolicyElasticData elasticData = new PolicyElasticData(policyData);
 340         JSONObject jsonObj = new JSONObject(elasticData);
 341         Index elkPut = new Index.Builder(jsonObj.toString()).
 342                 index(ELK_INDEX_POLICY).
 343                 type(indexType.name()).
 344                 id(elasticData.getPolicyName()).
 345                 refresh(true).
 346                 build();
 347
 348         JestResult result = jestClient.execute(elkPut);
 349
 350         if (result.isSucceeded()) {
 351             if (LOGGER.isInfoEnabled())
 352                 LOGGER.info("ElkConnector: OK: PUT operation of " + "->"  + ": " +
 353                         "success=" + result.isSucceeded() + "[" + result.getResponseCode() + ":" +
 354                         result.getPathToResult() + "]" + System.lineSeparator() +
 355                         result.getJsonString());
 356         } else {
 357             if (LOGGER.isWarnEnabled())
 358                 LOGGER.warn("ElkConnector: FAILURE: PUT operation of "+ "->" + ": " +
 359                         "success=" + result.isSucceeded() + "[" + result.getResponseCode() + ":" +
 360                         result.getPathToResult() + "]" + System.lineSeparator() +
 361                         result.getJsonString());
 362
 363         }
 364
 365         return result.isSucceeded();
 366     }
 367
 368     @Override
 369     public boolean delete(PolicyRestAdapter policyData) throws IllegalStateException  {
 370         PolicyIndexType indexType = null;
 371         JestResult result;
 372         try {
 373             String policyName = policyData.getNewFileName();
 374             if(policyName.contains("Config_")){
 375                 policyName = policyName.replace(".Config_", ":Config_");
 376             }else if(policyName.contains("Action_")){
 377                 policyName = policyName.replace(".Action_", ":Action_");
 378             }else if(policyName.contains("Decision_")){
 379                 policyName = policyName.replace(".Decision_", ":Decision_");
 380             }
 381
 382             String[] splitPolicyName = policyName.split(":");
 383             indexType = ElkConnector.toPolicyIndexType(splitPolicyName[1]);
 384             if (!isType(indexType)) {
 385                 throw new IllegalStateException("ELK: Index: " + ELK_INDEX_POLICY +
 386                         " Type: " + indexType +
 387                         " is not configured");
 388             }
 389             PolicyElasticData elasticData = new PolicyElasticData(policyData);
 390             Delete deleteRequest = new Delete.Builder(elasticData.getPolicyName()).index(ELK_INDEX_POLICY).
 391                     type(indexType.name()).build();
 392             result = jestClient.execute(deleteRequest);
 393         } catch (IllegalArgumentException | IOException e) {
 394             LOGGER.warn(XACMLErrorConstants.ERROR_SYSTEM_ERROR + ": delete:" +
 395                     indexType +  ": null" + ":" + policyData.getNewFileName() + ": " +
 396                     e.getMessage(), e);
 397             throw new IllegalStateException(e);
 398         }
 399
 400         if (result.isSucceeded()) {
 401             if (LOGGER.isInfoEnabled())
 402                 LOGGER.info("OK: DELETE operation of " + indexType + ":" + policyData.getNewFileName() + ": " +
 403                         "success=" + result.isSucceeded() + "[" + result.getResponseCode() + ":" +
 404                         result.getPathToResult() + "]" + System.lineSeparator() +
 405                         result.getJsonString());
 406         } else {
 407             if (LOGGER.isWarnEnabled())
 408                 LOGGER.warn("FAILURE: DELETE operation of " + indexType + ":" + policyData.getNewFileName() + ": " +
 409                         "success=" + result.isSucceeded() + "[" + result.getResponseCode() + ":" +
 410                         result.getPathToResult() + "]" + System.lineSeparator() +
 411                         result.getJsonString());
 412         }
 413
 414         return result.isSucceeded();
 415     }
 416
 417     @Override
 418     public boolean update(PolicyRestAdapter policyData) throws IllegalStateException  {
 419         if (LOGGER.isDebugEnabled()){
 420             LOGGER.debug("ENTER");
 421         }
 422         try {
 423             boolean success = put(policyData);
 424             return success;
 425         } catch (Exception e) {
 426             LOGGER.warn(XACMLErrorConstants.ERROR_UNKNOWN + ":" + "cannot test and update", e);
 427             throw new IllegalStateException(e);
 428         }
 429     }
 430 }