From 3fdfdf3b83a1d6d80066c321eb3f05568657aba7 Mon Sep 17 00:00:00 2001 From: Gao Hongtao Date: Sun, 19 Jul 2020 23:37:13 +0800 Subject: [PATCH] Storage elasticsearch health check (#5099) --- .../skywalking/apm/util/StringUtil.java | 8 ++ .../server-library/library-client/pom.xml | 6 + .../elasticsearch/ElasticSearchClient.java | 125 ++++++++++++++---- .../healthcheck/DelegatedHealthChecker.java | 39 ++++++ .../client/healthcheck/HealthCheckable.java | 34 +++++ .../jdbc/hikaricp/JDBCHikariCPClient.java | 34 ++--- .../server/library/util/HealthChecker.java | 37 ++++++ .../StorageModuleElasticsearchProvider.java | 7 + .../StorageModuleElasticsearch7Provider.java | 7 + .../client/ElasticSearch7Client.java | 89 +++++++++---- .../plugin/jdbc/h2/H2StorageProvider.java | 13 +- .../telemetry/api/HealthCheckMetrics.java | 45 +++++++ .../server/telemetry/api/MetricsCreator.java | 6 +- 13 files changed, 368 insertions(+), 82 deletions(-) create mode 100644 oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/DelegatedHealthChecker.java create mode 100644 oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/HealthCheckable.java create mode 100644 oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/HealthChecker.java create mode 100644 oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HealthCheckMetrics.java diff --git a/apm-commons/apm-util/src/main/java/org/apache/skywalking/apm/util/StringUtil.java b/apm-commons/apm-util/src/main/java/org/apache/skywalking/apm/util/StringUtil.java index afbadd9e3..8cd72a816 100644 --- a/apm-commons/apm-util/src/main/java/org/apache/skywalking/apm/util/StringUtil.java +++ b/apm-commons/apm-util/src/main/java/org/apache/skywalking/apm/util/StringUtil.java @@ -18,6 +18,8 @@ package org.apache.skywalking.apm.util; +import java.util.function.Consumer; + public final class StringUtil { public static boolean isEmpty(String str) { return str == null || str.length() == 0; @@ -27,6 +29,12 @@ public final class StringUtil { return !isEmpty(str); } + public static void setIfPresent(String value, Consumer setter) { + if (isNotEmpty(value)) { + setter.accept(value); + } + } + public static String join(final char delimiter, final String... strings) { if (strings.length == 0) { return null; diff --git a/oap-server/server-library/library-client/pom.xml b/oap-server/server-library/library-client/pom.xml index 39152b2c0..967e1fbc5 100755 --- a/oap-server/server-library/library-client/pom.xml +++ b/oap-server/server-library/library-client/pom.xml @@ -29,6 +29,12 @@ jar + + org.apache.skywalking + library-util + ${project.version} + + io.grpc grpc-core diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java index 5d2a90ffb..5059caf63 100644 --- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java @@ -37,7 +37,9 @@ import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.concurrent.locks.ReentrantLock; import javax.net.ssl.SSLContext; +import lombok.RequiredArgsConstructor; import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpEntity; @@ -58,8 +60,11 @@ import org.apache.http.ssl.SSLContextBuilder; import org.apache.http.ssl.SSLContexts; import org.apache.skywalking.apm.util.StringUtil; import org.apache.skywalking.oap.server.library.client.Client; +import org.apache.skywalking.oap.server.library.client.healthcheck.DelegatedHealthChecker; +import org.apache.skywalking.oap.server.library.client.healthcheck.HealthCheckable; import org.apache.skywalking.oap.server.library.client.request.InsertRequest; import org.apache.skywalking.oap.server.library.client.request.UpdateRequest; +import org.apache.skywalking.oap.server.library.util.HealthChecker; import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; @@ -90,7 +95,8 @@ import org.elasticsearch.search.builder.SearchSourceBuilder; * ElasticSearchClient connects to the ES server by using ES client APIs. */ @Slf4j -public class ElasticSearchClient implements Client { +@RequiredArgsConstructor +public class ElasticSearchClient implements Client, HealthCheckable { public static final String TYPE = "type"; protected final String clusterNodes; protected final String protocol; @@ -103,6 +109,8 @@ public class ElasticSearchClient implements Client { private volatile String password; private final List indexNameConverters; protected volatile RestHighLevelClient client; + protected DelegatedHealthChecker healthChecker = new DelegatedHealthChecker(); + protected final ReentrantLock connectLock = new ReentrantLock(); public ElasticSearchClient(String clusterNodes, String protocol, @@ -122,16 +130,21 @@ public class ElasticSearchClient implements Client { @Override public void connect() throws IOException, KeyStoreException, NoSuchAlgorithmException, KeyManagementException, CertificateException { - List hosts = parseClusterNodes(protocol, clusterNodes); - if (client != null) { - try { - client.close(); - } catch (Throwable t) { - log.error("ElasticSearch client reconnection fails based on new config", t); + connectLock.lock(); + try { + List hosts = parseClusterNodes(protocol, clusterNodes); + if (client != null) { + try { + client.close(); + } catch (Throwable t) { + log.error("ElasticSearch client reconnection fails based on new config", t); + } } + client = createClient(hosts); + client.ping(); + } finally { + connectLock.unlock(); } - client = createClient(hosts); - client.ping(); } protected RestHighLevelClient createClient( @@ -208,10 +221,23 @@ public class ElasticSearchClient implements Client { public List retrievalIndexByAliases(String aliases) throws IOException { aliases = formatIndexName(aliases); - Response response = client.getLowLevelClient().performRequest(HttpGet.METHOD_NAME, "/_alias/" + aliases); + Response response; + try { + response = client.getLowLevelClient().performRequest(HttpGet.METHOD_NAME, "/_alias/" + aliases); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } if (HttpStatus.SC_OK == response.getStatusLine().getStatusCode()) { Gson gson = new Gson(); - InputStreamReader reader = new InputStreamReader(response.getEntity().getContent()); + InputStreamReader reader; + try { + reader = new InputStreamReader(response.getEntity().getContent()); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } JsonObject responseJson = gson.fromJson(reader, JsonObject.class); log.debug("retrieval indexes by aliases {}, response is {}", aliases, responseJson); return new ArrayList<>(responseJson.keySet()); @@ -310,13 +336,44 @@ public class ElasticSearchClient implements Client { SearchRequest searchRequest = new SearchRequest(indexName); searchRequest.types(TYPE); searchRequest.source(searchSourceBuilder); - return client.search(searchRequest); + try { + SearchResponse response = client.search(searchRequest); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + handleIOPoolStopped(t); + throw t; + } + } + + protected void handleIOPoolStopped(Throwable t) throws IOException { + if (!(t instanceof IllegalStateException)) { + return; + } + IllegalStateException ise = (IllegalStateException) t; + // Fixed the issue described in https://github.com/elastic/elasticsearch/issues/39946 + if (ise.getMessage().contains("I/O reactor status: STOPPED") && + connectLock.tryLock()) { + try { + connect(); + } catch (KeyStoreException | NoSuchAlgorithmException | KeyManagementException | CertificateException e) { + throw new IllegalStateException("Can't reconnect to Elasticsearch", e); + } + } } public GetResponse get(String indexName, String id) throws IOException { indexName = formatIndexName(indexName); GetRequest request = new GetRequest(indexName, TYPE, id); - return client.get(request); + try { + GetResponse response = client.get(request); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public SearchResponse ids(String indexName, String[] ids) throws IOException { @@ -325,28 +382,39 @@ public class ElasticSearchClient implements Client { SearchRequest searchRequest = new SearchRequest(indexName); searchRequest.types(TYPE); searchRequest.source().query(QueryBuilders.idsQuery().addIds(ids)).size(ids.length); - return client.search(searchRequest); + try { + SearchResponse response = client.search(searchRequest); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public void forceInsert(String indexName, String id, XContentBuilder source) throws IOException { IndexRequest request = (IndexRequest) prepareInsert(indexName, id, source); request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.index(request); - } - - public void forceUpdate(String indexName, String id, XContentBuilder source, long version) throws IOException { - org.elasticsearch.action.update.UpdateRequest request = (org.elasticsearch.action.update.UpdateRequest) prepareUpdate( - indexName, id, source); - request.version(version); - request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.update(request); + try { + client.index(request); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public void forceUpdate(String indexName, String id, XContentBuilder source) throws IOException { org.elasticsearch.action.update.UpdateRequest request = (org.elasticsearch.action.update.UpdateRequest) prepareUpdate( indexName, id, source); request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.update(request); + try { + client.update(request); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public InsertRequest prepareInsert(String indexName, String id, XContentBuilder source) { @@ -379,8 +447,9 @@ public class ElasticSearchClient implements Client { int size = request.requests().size(); BulkResponse responses = client.bulk(request); log.info("Synchronous bulk took time: {} millis, size: {}", responses.getTook().getMillis(), size); - } catch (IOException e) { - log.error(e.getMessage(), e); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); } } @@ -431,4 +500,8 @@ public class ElasticSearchClient implements Client { } return indexName; } + + @Override public void registerChecker(HealthChecker healthChecker) { + this.healthChecker.register(healthChecker); + } } diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/DelegatedHealthChecker.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/DelegatedHealthChecker.java new file mode 100644 index 000000000..b386b3a18 --- /dev/null +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/DelegatedHealthChecker.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.library.client.healthcheck; + +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.skywalking.oap.server.library.util.HealthChecker; + +public class DelegatedHealthChecker implements HealthChecker { + private final AtomicReference delegated = new AtomicReference<>(); + + @Override public void health() { + Optional.ofNullable(delegated.get()).ifPresent(HealthChecker::health); + } + + @Override public void unHealth(Throwable t) { + Optional.ofNullable(delegated.get()).ifPresent(d -> d.unHealth(t)); + } + + public void register(HealthChecker healthChecker) { + delegated.set(healthChecker); + } +} diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/HealthCheckable.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/HealthCheckable.java new file mode 100644 index 000000000..6b1563185 --- /dev/null +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/healthcheck/HealthCheckable.java @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.library.client.healthcheck; + +import org.apache.skywalking.oap.server.library.util.HealthChecker; + +/** + * HealthCheckable indicate the client has the capacity of health check and need to register healthChecker. + */ +public interface HealthCheckable { + + /** + * Register health checker. + * + * @param healthChecker HealthChecker to be registered. + */ + void registerChecker(HealthChecker healthChecker); +} diff --git a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java index f99a6a052..d02ef4606 100644 --- a/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java +++ b/oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/jdbc/hikaricp/JDBCHikariCPClient.java @@ -26,37 +26,27 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.Properties; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; import org.apache.skywalking.oap.server.library.client.Client; +import org.apache.skywalking.oap.server.library.client.healthcheck.DelegatedHealthChecker; +import org.apache.skywalking.oap.server.library.client.healthcheck.HealthCheckable; import org.apache.skywalking.oap.server.library.client.jdbc.JDBCClientException; +import org.apache.skywalking.oap.server.library.util.HealthChecker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * JDBC Client uses HikariCP connection management lib to execute SQL. */ -public class JDBCHikariCPClient implements Client { +public class JDBCHikariCPClient implements Client, HealthCheckable { private static final Logger logger = LoggerFactory.getLogger(JDBCHikariCPClient.class); + private final HikariConfig hikariConfig; + private final DelegatedHealthChecker healthChecker; private HikariDataSource dataSource; - private HikariConfig hikariConfig; public JDBCHikariCPClient(Properties properties) { hikariConfig = new HikariConfig(properties); - } - - public void setHealthCheckListener(Consumer healthListener) { - ScheduledExecutorService asyncHealthScheduler = Executors.newSingleThreadScheduledExecutor(); - asyncHealthScheduler.scheduleAtFixedRate(() -> { - try (Connection c = dataSource.getConnection()) { - healthListener.accept(true); - } catch (SQLException ignored) { - healthListener.accept(false); - } - }, 0, 3, TimeUnit.SECONDS); + this.healthChecker = new DelegatedHealthChecker(); } @Override @@ -93,7 +83,9 @@ public class JDBCHikariCPClient implements Client { logger.debug("execute aql: {}", sql); try (Statement statement = connection.createStatement()) { statement.execute(sql); + healthChecker.health(); } catch (SQLException e) { + healthChecker.unHealth(e); throw new JDBCClientException(e.getMessage(), e); } } @@ -107,6 +99,7 @@ public class JDBCHikariCPClient implements Client { setStatementParam(statement, params); result = statement.execute(); statement.closeOnCompletion(); + healthChecker.health(); } catch (SQLException e) { if (statement != null) { try { @@ -114,6 +107,7 @@ public class JDBCHikariCPClient implements Client { } catch (SQLException e1) { } } + healthChecker.unHealth(e); throw new JDBCClientException(e.getMessage(), e); } @@ -129,6 +123,7 @@ public class JDBCHikariCPClient implements Client { setStatementParam(statement, params); rs = statement.executeQuery(); statement.closeOnCompletion(); + healthChecker.health(); } catch (SQLException e) { if (statement != null) { try { @@ -136,6 +131,7 @@ public class JDBCHikariCPClient implements Client { } catch (SQLException e1) { } } + healthChecker.unHealth(e); throw new JDBCClientException(e.getMessage(), e); } @@ -161,4 +157,8 @@ public class JDBCHikariCPClient implements Client { } } } + + @Override public void registerChecker(HealthChecker healthChecker) { + this.healthChecker.register(healthChecker); + } } diff --git a/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/HealthChecker.java b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/HealthChecker.java new file mode 100644 index 000000000..809a05f33 --- /dev/null +++ b/oap-server/server-library/library-util/src/main/java/org/apache/skywalking/oap/server/library/util/HealthChecker.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.library.util; + +/** + * Health checker provides methods to register the health status. + */ +public interface HealthChecker { + + /** + * It's health. + */ + void health(); + + /** + * It's unHealth. + * + * @param t details of unhealthy status + */ + void unHealth(Throwable t); +} diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java index 9f591cbc8..75f17f722 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/StorageModuleElasticsearchProvider.java @@ -74,6 +74,10 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopNR import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TopologyQueryEsDAO; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.TraceQueryEsDAO; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query.UITemplateManagementEsDAO; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics; +import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator; +import org.apache.skywalking.oap.server.telemetry.api.MetricsTag; /** * The storage provider for ElasticSearch 6. @@ -181,6 +185,9 @@ public class StorageModuleElasticsearchProvider extends ModuleProvider { @Override public void start() throws ModuleStartException { + MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME).provider().getService(MetricsCreator.class); + HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge("storage_elasticsearch", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); + elasticSearchClient.registerChecker(healthChecker); try { elasticSearchClient.connect(); StorageEsInstaller installer = new StorageEsInstaller(elasticSearchClient, getManager(), config); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java index c69d399cb..98da53cc7 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/StorageModuleElasticsearch7Provider.java @@ -70,6 +70,10 @@ import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.Meta import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.MetricsQueryEs7DAO; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.ProfileThreadSnapshotQueryEs7DAO; import org.apache.skywalking.oap.server.storage.plugin.elasticsearch7.query.TraceQueryEs7DAO; +import org.apache.skywalking.oap.server.telemetry.TelemetryModule; +import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics; +import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator; +import org.apache.skywalking.oap.server.telemetry.api.MetricsTag; import static org.apache.skywalking.oap.server.storage.plugin.elasticsearch.StorageModuleElasticsearchProvider.indexNameConverters; @@ -186,6 +190,9 @@ public class StorageModuleElasticsearch7Provider extends ModuleProvider { @Override public void start() throws ModuleStartException { + MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME).provider().getService(MetricsCreator.class); + HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge("storage_elasticsearch", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); + elasticSearch7Client.registerChecker(healthChecker); try { elasticSearch7Client.connect(); diff --git a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java index e561c826c..b1534b0a3 100644 --- a/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java +++ b/oap-server/server-storage-plugin/storage-elasticsearch7-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch7/client/ElasticSearch7Client.java @@ -84,16 +84,21 @@ public class ElasticSearch7Client extends ElasticSearchClient { @Override public void connect() throws IOException, KeyStoreException, NoSuchAlgorithmException, KeyManagementException, CertificateException { - if (client != null) { - try { - client.close(); - } catch (Throwable t) { - log.error("ElasticSearch7 client reconnection fails based on new config", t); + connectLock.lock(); + try { + if (client != null) { + try { + client.close(); + } catch (Throwable t) { + log.error("ElasticSearch7 client reconnection fails based on new config", t); + } } + List hosts = parseClusterNodes(protocol, clusterNodes); + client = createClient(hosts); + client.ping(RequestOptions.DEFAULT); + } finally { + connectLock.unlock(); } - List hosts = parseClusterNodes(protocol, clusterNodes); - client = createClient(hosts); - client.ping(RequestOptions.DEFAULT); } public boolean createIndex(String indexName) throws IOException { @@ -124,7 +129,14 @@ public class ElasticSearch7Client extends ElasticSearchClient { aliases = formatIndexName(aliases); GetAliasesRequest getAliasesRequest = new GetAliasesRequest(aliases); - GetAliasesResponse alias = client.indices().getAlias(getAliasesRequest, RequestOptions.DEFAULT); + GetAliasesResponse alias; + try { + alias = client.indices().getAlias(getAliasesRequest, RequestOptions.DEFAULT); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } return new ArrayList<>(alias.getAliases().keySet()); } @@ -184,13 +196,28 @@ public class ElasticSearch7Client extends ElasticSearchClient { indexName = formatIndexName(indexName); SearchRequest searchRequest = new SearchRequest(indexName); searchRequest.source(searchSourceBuilder); - return client.search(searchRequest, RequestOptions.DEFAULT); + try { + SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + handleIOPoolStopped(t); + throw t; + } } public GetResponse get(String indexName, String id) throws IOException { indexName = formatIndexName(indexName); GetRequest request = new GetRequest(indexName, id); - return client.get(request, RequestOptions.DEFAULT); + try { + GetResponse response = client.get(request, RequestOptions.DEFAULT); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public SearchResponse ids(String indexName, String[] ids) throws IOException { @@ -198,30 +225,39 @@ public class ElasticSearch7Client extends ElasticSearchClient { SearchRequest searchRequest = new SearchRequest(indexName); searchRequest.source().query(QueryBuilders.idsQuery().addIds(ids)).size(ids.length); - return client.search(searchRequest, RequestOptions.DEFAULT); + try { + SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); + healthChecker.health(); + return response; + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public void forceInsert(String indexName, String id, XContentBuilder source) throws IOException { IndexRequest request = (IndexRequest) prepareInsert(indexName, id, source); request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.index(request, RequestOptions.DEFAULT); - } - - public void forceUpdate(String indexName, String id, XContentBuilder source, long seqNo, - long primaryTerm) throws IOException { - org.elasticsearch.action.update.UpdateRequest request = (org.elasticsearch.action.update.UpdateRequest) prepareUpdate( - indexName, id, source); - request.setIfSeqNo(seqNo); - request.setIfPrimaryTerm(primaryTerm); - request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.update(request, RequestOptions.DEFAULT); + try { + client.index(request, RequestOptions.DEFAULT); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public void forceUpdate(String indexName, String id, XContentBuilder source) throws IOException { org.elasticsearch.action.update.UpdateRequest request = (org.elasticsearch.action.update.UpdateRequest) prepareUpdate( indexName, id, source); request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); - client.update(request, RequestOptions.DEFAULT); + try { + client.update(request, RequestOptions.DEFAULT); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); + throw t; + } } public InsertRequest prepareInsert(String indexName, String id, XContentBuilder source) { @@ -256,8 +292,9 @@ public class ElasticSearch7Client extends ElasticSearchClient { int size = request.requests().size(); BulkResponse responses = client.bulk(request, RequestOptions.DEFAULT); log.info("Synchronous bulk took time: {} millis, size: {}", responses.getTook().getMillis(), size); - } catch (IOException e) { - log.error(e.getMessage(), e); + healthChecker.health(); + } catch (Throwable t) { + healthChecker.unHealth(t); } } diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java index 067bf2ce8..265352268 100644 --- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java +++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/h2/H2StorageProvider.java @@ -64,7 +64,7 @@ import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TopologyQue import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2TraceQueryDAO; import org.apache.skywalking.oap.server.storage.plugin.jdbc.h2.dao.H2UITemplateManagementDAO; import org.apache.skywalking.oap.server.telemetry.TelemetryModule; -import org.apache.skywalking.oap.server.telemetry.api.GaugeMetrics; +import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics; import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator; import org.apache.skywalking.oap.server.telemetry.api.MetricsTag; @@ -136,8 +136,8 @@ public class H2StorageProvider extends ModuleProvider { @Override public void start() throws ServiceNotProvidedException, ModuleStartException { MetricsCreator metricCreator = getManager().find(TelemetryModule.NAME).provider().getService(MetricsCreator.class); - GaugeMetrics healthChecker = metricCreator.createHealthCheckerGauge("storage_h2", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); - healthChecker.setValue(1); + HealthCheckMetrics healthChecker = metricCreator.createHealthCheckerGauge("storage_h2", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE); + h2Client.registerChecker(healthChecker); try { h2Client.connect(); @@ -146,13 +146,6 @@ public class H2StorageProvider extends ModuleProvider { } catch (StorageException e) { throw new ModuleStartException(e.getMessage(), e); } - h2Client.setHealthCheckListener(isHealthy -> { - if (isHealthy) { - healthChecker.setValue(0); - } else { - healthChecker.setValue(1); - } - }); } @Override diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HealthCheckMetrics.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HealthCheckMetrics.java new file mode 100644 index 000000000..ccb2e91c4 --- /dev/null +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/HealthCheckMetrics.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.oap.server.telemetry.api; + +import lombok.extern.slf4j.Slf4j; +import org.apache.skywalking.oap.server.library.util.HealthChecker; + +/** + * HealthCheckMetrics intends to record health status. + */ +@Slf4j +public class HealthCheckMetrics implements HealthChecker { + private final GaugeMetrics metrics; + + public HealthCheckMetrics(GaugeMetrics metrics) { + this.metrics = metrics; + // The initial status is unhealthy with -1 code. + metrics.setValue(-1); + } + + public void health() { + metrics.setValue(0); + } + + public void unHealth(Throwable t) { + log.error("Health check fails", t); + metrics.setValue(1); + } +} diff --git a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCreator.java b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCreator.java index 11d14ff9f..5bd89d26a 100644 --- a/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCreator.java +++ b/oap-server/server-telemetry/telemetry-api/src/main/java/org/apache/skywalking/oap/server/telemetry/api/MetricsCreator.java @@ -50,11 +50,11 @@ public interface MetricsCreator extends Service { /** * Create a Health Check gauge. */ - default GaugeMetrics createHealthCheckerGauge(String name, MetricsTag.Keys tagKeys, MetricsTag.Values tagValues) { + default HealthCheckMetrics createHealthCheckerGauge(String name, MetricsTag.Keys tagKeys, MetricsTag.Values tagValues) { Preconditions.checkArgument(!Strings.isNullOrEmpty(name), "Require non-null or empty metric name"); - return createGauge(Strings.lenientFormat("%s%s", HEALTH_METRIC_PREFIX, name), + return new HealthCheckMetrics(createGauge(Strings.lenientFormat("%s%s", HEALTH_METRIC_PREFIX, name), Strings.lenientFormat("%s health check", name), - tagKeys, tagValues); + tagKeys, tagValues)); } /**