From f18883b4aa42334c39b523ead45ecb18ed2f5e92 Mon Sep 17 00:00:00 2001 From: ZhangZihao Date: Wed, 24 Apr 2019 16:59:34 +0800 Subject: ‘correlateClearedMessage’ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change-Id: I32ea4043d32f29c920b370de3c84341c218ed5c2 Issue-ID: DCAEGEN2-1190 Signed-off-by: ZhangZihao --- .../feeder/config/ApplicationConfiguration.java | 2 + .../feeder/service/ElasticsearchService.java | 115 ++++++++++++++++++--- 2 files changed, 100 insertions(+), 17 deletions(-) (limited to 'components/datalake-handler/feeder/src/main/java') diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java index 21ca1b70..a3add0ed 100644 --- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java +++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/config/ApplicationConfiguration.java @@ -57,4 +57,6 @@ public class ApplicationConfiguration { private String rawDataLabel; private String defaultTopicName; + + private String elasticsearchType; } diff --git a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/ElasticsearchService.java b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/ElasticsearchService.java index 2c16b2b8..30aa7332 100644 --- a/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/ElasticsearchService.java +++ b/components/datalake-handler/feeder/src/main/java/org/onap/datalake/feeder/service/ElasticsearchService.java @@ -26,8 +26,13 @@ import java.util.List; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; +import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpHost; +import org.elasticsearch.ElasticsearchException; import org.elasticsearch.action.ActionListener; +import org.elasticsearch.action.get.GetRequest; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.client.indices.CreateIndexRequest; import org.elasticsearch.client.indices.CreateIndexResponse; import org.elasticsearch.client.indices.GetIndexRequest; @@ -38,6 +43,7 @@ import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType; +import org.elasticsearch.rest.RestStatus; import org.json.JSONObject; import org.onap.datalake.feeder.config.ApplicationConfiguration; import org.onap.datalake.feeder.domain.Db; @@ -118,15 +124,15 @@ public class ElasticsearchService { for (JSONObject json : jsons) { if(topic.isCorrelateClearedMessage()) { - boolean found = correlateClearedMessage(json); + boolean found = correlateClearedMessage(topic, json); if(found) { continue; } } String id = topic.getMessageId(json); //id can be null - - request.add(new IndexRequest(topic.getName().toLowerCase(), "doc", id).source(json.toString(), XContentType.JSON)); + + request.add(new IndexRequest(topic.getName().toLowerCase(), config.getElasticsearchType(), id).source(json.toString(), XContentType.JSON)); } if(config.isAsync()) { client.bulkAsync(request, RequestOptions.DEFAULT, listener); @@ -138,21 +144,96 @@ public class ElasticsearchService { } } } - - private boolean correlateClearedMessage(JSONObject json) { - boolean found = true; - - /*TODO - * 1. check if this is a alarm cleared message - * 2. search previous alarm message - * 3. update previous message, if success, set found=true - */ - //for Sonar test, remove the following - if(json.isNull("kkkkk")) { - found = false; + + /** + * + * @param topic + * @param json + * @return boolean + * + * Because of query by id, The search API cannot be used for query. + * The search API can only query all data or based on the fields in the source. + * So use the get API, three parameters: index, type, document id + */ + private boolean correlateClearedMessage(Topic topic, JSONObject json) { + boolean found = false; + String eName = null; + + try { + eName = json.query("/event/commonEventHeader/eventName").toString(); + + if (StringUtils.isNotBlank(eName)) { + + if (eName.endsWith("Cleared")) { + + String name = eName.substring(0, eName.length() - 7); + String reportingEntityName = json.query("/event/commonEventHeader/reportingEntityName").toString(); + String specificProblem = json.query("/event/faultFields/specificProblem").toString(); + + String id = null; + StringBuilder stringBuilder = new StringBuilder(); + stringBuilder = stringBuilder.append(name).append('^').append(reportingEntityName).append('^').append(specificProblem); + + id = stringBuilder.toString();//example: id = "aaaa^cccc^bbbbb" + String index = topic.getName().toLowerCase(); + + //get + GetRequest getRequest = new GetRequest(index, config.getElasticsearchType(), id); + + GetResponse getResponse = null; + try { + getResponse = client.get(getRequest, RequestOptions.DEFAULT); + if (getResponse != null) { + + if (getResponse.isExists()) { + String sourceAsString = getResponse.getSourceAsString(); + JSONObject jsonObject = new JSONObject(sourceAsString); + jsonObject.getJSONObject("event").getJSONObject("faultFields").put("vfStatus", "closed"); + String jsonString = jsonObject.toString(); + + //update + IndexRequest request = new IndexRequest(index, config.getElasticsearchType(), id); + request.source(jsonString, XContentType.JSON); + IndexResponse indexResponse = null; + try { + indexResponse = client.index(request, RequestOptions.DEFAULT); + found = true; + } catch (IOException e) { + log.error("save failure"); + } + } else { + log.error("The getResponse was not exists" ); + } + + } else { + log.error("The document for this id was not found" ); + } + + } catch (ElasticsearchException e) { + if (e.status() == RestStatus.NOT_FOUND) { + log.error("The document for this id was not found" ); + } + if (e.status() == RestStatus.CONFLICT) { + log.error("Version conflict" ); + } + log.error("Get document exception", e); + }catch (IOException e) { + log.error(topic.getName() , e); + } + + } else { + log.info("The data is normal"); + } + + } else { + log.debug("event id null"); + } + + } catch (Exception e) { + log.error("error",e); } - - return found; + + return found; } } -- cgit 1.2.3-korg