2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 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
11 * http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
21 package org.openecomp.sdc.be.dao.impl;
23 import com.fasterxml.jackson.core.JsonProcessingException;
24 import fj.data.Either;
25 import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
26 import org.elasticsearch.cluster.health.ClusterHealthStatus;
27 import org.elasticsearch.common.unit.TimeValue;
28 import org.openecomp.sdc.be.config.BeEcompErrorManager;
29 import org.openecomp.sdc.be.config.ConfigurationManager;
30 import org.openecomp.sdc.be.dao.api.ESGenericSearchDAO;
31 import org.openecomp.sdc.be.dao.api.ICatalogDAO;
32 import org.openecomp.sdc.be.dao.api.ResourceUploadStatus;
33 import org.openecomp.sdc.be.resources.data.ESArtifactData;
34 import org.openecomp.sdc.be.resources.exception.ResourceDAOException;
35 import org.openecomp.sdc.common.api.HealthCheckInfo.HealthCheckStatus;
36 import org.openecomp.sdc.common.log.wrappers.Logger;
37 import org.springframework.stereotype.Component;
39 import javax.annotation.PostConstruct;
40 import java.util.List;
41 import java.util.concurrent.Executors;
42 import java.util.concurrent.ScheduledExecutorService;
43 import java.util.concurrent.ThreadFactory;
44 import java.util.concurrent.TimeUnit;
46 @Component("resource-dao")
47 public class ESCatalogDAO extends ESGenericSearchDAO implements ICatalogDAO {
49 private static Logger log = Logger.getLogger(ESCatalogDAO.class.getName());
51 // Index Checking Variables
52 private boolean initCompleted = false;
54 //TODO use LoggerMetric instead
55 private static Logger healthCheckLogger = Logger.getLogger("elasticsearch.healthcheck");
57 ///// HealthCheck/////////
58 private static final String ES_HEALTH_CHECK_STR = "elasticsearchHealthCheck";
60 private ScheduledExecutorService healthCheckScheduler = Executors
61 .newSingleThreadScheduledExecutor(new ThreadFactory() {
63 public Thread newThread(Runnable r) {
64 return new Thread(r, "ES-Health-Check-Thread");
68 private class HealthCheckScheduledTask implements Runnable {
71 log.trace("Executing ELASTICSEARCH Health Check Task - Start");
73 HealthCheckStatus healthStatus = null;
75 healthStatus = isInitCompleted() ? checkHealth() : HealthCheckStatus.DOWN;
76 } catch (Exception e) {
77 log.error("Error while trying to connect to elasticsearch. host: {} | port: {} | error: {}",
78 getEsClient().getServerHost(), getEsClient().getServerPort(), e.getMessage(), e);
79 healthStatus = HealthCheckStatus.DOWN;
81 log.trace("Executed ELASTICSEARCH Health Check Task - Status = {}", healthStatus);
82 if (healthStatus != lastHealthState) {
83 log.trace("ELASTICSEARCH Health State Changed to {}. Issuing alarm / recovery alarm...", healthStatus);
84 lastHealthState = healthStatus;
90 private HealthCheckScheduledTask healthCheckScheduledTask = new HealthCheckScheduledTask();
91 private volatile HealthCheckStatus lastHealthState = HealthCheckStatus.DOWN;
94 * Get ES cluster status string rep
96 * @return "GREEN", "YELLOW" or "RED"
98 private HealthCheckStatus checkHealth() {
99 if (!isInitCompleted()) {
100 return HealthCheckStatus.DOWN;
102 ClusterHealthRequest healthRequest = new ClusterHealthRequest("_all");
103 healthRequest.masterNodeTimeout(TimeValue.timeValueSeconds(2));
104 ClusterHealthStatus status = getClient().admin().cluster().health(healthRequest).actionGet().getStatus();
105 healthCheckLogger.debug("ES cluster health status is {}", status);
106 if (status == null || status.equals(ClusterHealthStatus.RED)) {
107 return HealthCheckStatus.DOWN;
109 return HealthCheckStatus.UP;
112 private void logAlarm() {
113 if (lastHealthState == HealthCheckStatus.UP) {
114 BeEcompErrorManager.getInstance().logBeHealthCheckElasticSearchRecovery(ES_HEALTH_CHECK_STR);
116 BeEcompErrorManager.getInstance().logBeHealthCheckElasticSearchError(ES_HEALTH_CHECK_STR);
121 public void initCompleted() {
122 long interval = ConfigurationManager.getConfigurationManager().getConfiguration()
123 .getEsReconnectIntervalInSeconds(5);
124 this.healthCheckScheduler.scheduleAtFixedRate(healthCheckScheduledTask, 0, interval, TimeUnit.SECONDS);
125 initCompleted = true;
129 public void writeArtifact(ESArtifactData artifactData) {
131 saveResourceData(artifactData);
132 } catch (Exception e) {
133 throw new ResourceDAOException("Error to save ArtifactData with " + artifactData.getId());
138 public Either<ESArtifactData, ResourceUploadStatus> getArtifact(String id) {
139 ESArtifactData resData = null;
142 resData = findById(getTypeFromClass(ESArtifactData.class), id, ESArtifactData.class);
143 } catch (Exception e) {
145 BeEcompErrorManager.getInstance().logBeDaoSystemError("Get Artifact from database");
146 log.debug("ESCatalogDAO:getArtifact failed with exception ", e);
147 return Either.right(ResourceUploadStatus.ERROR);
150 if (resData != null) {
151 return Either.left(resData);
153 return Either.right(ResourceUploadStatus.NOT_EXIST);
157 private <T> String getTypeFromClass(Class<T> clazz) {
159 return clazz.getSimpleName().toLowerCase();
163 public void deleteArtifact(String id) {
164 delete(getTypeFromClass(ESArtifactData.class), id);
168 public Either<List<ESArtifactData>, ResourceUploadStatus> getArtifacts(String[] ids) {
169 List<ESArtifactData> resData = null;
171 resData = findByIds(getTypeFromClass(ESArtifactData.class), ESArtifactData.class, ids);
172 } catch (Exception e) {
174 return Either.right(ResourceUploadStatus.ERROR);
177 if (resData != null && !resData.isEmpty()) {
178 return Either.left(resData);
180 return Either.right(ResourceUploadStatus.NOT_EXIST);
184 private void saveResourceData(ESArtifactData data) throws JsonProcessingException {
185 String typeName = getTypeFromClass(data.getClass());
186 saveResourceData(typeName, data, data.getId());
190 public void deleteAllArtifacts() {
191 String typeName = getTypeFromClass(ESArtifactData.class);
192 String indexName = getIndexForType(typeName);
193 deleteIndex(indexName);
197 public boolean isInitCompleted() {
198 return initCompleted;
201 public HealthCheckStatus getHealth() {
202 return lastHealthState;