Self Observability: add BanyanDB and Elasticsearch write latency metrics and dashboards (#13544)

Author: wankai123

docs/en/changes/changes.md

@@ -106,6 +106,7 @@
 * BanyanDB: fix when setting `@BanyanDB.TimestampColumn`, the column should not be indexed.
 * OAP Self Observability: make Trace analysis metrics separate by label `protocol`, add Zipkin span dropped metrics.
 * BanyanDB: Move data write logic from BanyanDB Java Client to OAP and support observe metrics for write operations.
+* Self Observability: add write latency metrics for BanyanDB and ElasticSearch.
 
 #### UI
 
@@ -137,6 +138,7 @@
 * Adapt new trace protocol and implement new trace view.
 * Implement Trace page.
 * Support collapsing and expanding for the event widget.
+* UI-template: add BanyanDB and Elasticsearch write latency dashboards for OAP self observability.
 
 #### Documentation
 
@@ -148,6 +150,7 @@
 * Enhance the async-profiling duration options.
 * Enhance the TTL Tab on Setting page.
 * Fix the snapshot charts in alarm page.
+* Fix `Fluent Bit` dead links.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/230?closed=1)
 

docs/en/setup/backend/backend-alarm.md

@@ -309,7 +309,7 @@ slack:
 ```
 
 ### WeChat
-Note that only the WeChat Company Edition (WeCom) supports WebHooks. To use the WeChat WebHook, follow the [Wechat Webhooks guide](https://work.weixin.qq.com/help?doc_id=13376).
+Note that only the WeChat Company Edition (WeCom) supports WebHooks. To use the WeChat WebHook, follow the [Wechat Webhooks guide](https://open.work.weixin.qq.com/help2/pc/14931).
 The alarm message will be sent through HTTP post by `application/json` content type after you have set up Wechat Webhooks as follows:
 ```yml
 wechat:

docs/en/setup/backend/backend-mysql-monitoring.md

@@ -45,7 +45,7 @@ SkyWalking leverages [fluentbit](https://fluentbit.io/) or other log agents for
 3. The SkyWalking OAP Server parses the expression with [LAL](../../concepts-and-designs/lal.md) to parse/extract and store the results.
 
 ### Set up
-1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/docker).
+1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/getting-started-with-fluent-bit).
 2. Config fluentbit from [here](../../../../test/e2e-v2/cases/mysql/mysql-slowsql/fluent-bit.conf) for MySQL or [here](../../../../test/e2e-v2/cases/mariadb/mariadb-slowsql/fluent-bit.conf) for MariaDB.
 3. Enable slow log from [here](../../../../test/e2e-v2/cases/mysql/mysql-slowsql/my.cnf) for MySQL or [here](../../../../test/e2e-v2/cases/mariadb/mariadb-slowsql/my.cnf) for MariaDB.
 

docs/en/setup/backend/backend-nginx-monitoring.md

@@ -121,7 +121,7 @@ SkyWalking leverages [fluentbit](https://fluentbit.io/) or other log agents for
 3. The SkyWalking OAP Server parses the expression with [LAL](../../concepts-and-designs/lal.md) to parse/extract and store the results.
 
 ### Set up
-1. Install [fluentbit](https://docs.fluentbit.io/manual/installation/docker).
+1. Install [fluentbit](https://docs.fluentbit.io/manual/installation/getting-started-with-fluent-bit).
 2. Config fluent bit with fluent-bit.conf, refer to [here](../../../../test/e2e-v2/cases/nginx/fluent-bit.conf).
 
 ### Error Log Monitoring

docs/en/setup/backend/backend-postgresql-monitoring.md

@@ -58,7 +58,7 @@ SkyWalking leverages [fluentbit](https://fluentbit.io/) or other log agents for
 3. The SkyWalking OAP Server parses the expression with [LAL](../../concepts-and-designs/lal.md) to parse/extract and store the results.
 
 ### Set up
-1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/docker).
+1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/getting-started-with-fluent-bit).
 2. Config [fluentbit](../../../../test/e2e-v2/cases/postgresql/postgres-exporter/fluent-bit.conf)
 3. Config PostgreSQL to enable slow log. [Example](../../../../test/e2e-v2/cases/postgresql/postgres-exporter/postgresql.conf).
 

docs/en/setup/backend/backend-redis-monitoring.md

@@ -46,7 +46,7 @@ SkyWalking leverages [fluentbit](https://fluentbit.io/) or other log agents for
 4. The SkyWalking OAP Server parses the expression with [LAL](../../concepts-and-designs/lal.md) to parse/extract and store the results.
 
 ### Set up
-1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/docker).
+1. Set up [fluentbit](https://docs.fluentbit.io/manual/installation/getting-started-with-fluent-bit).
 2. Config fluentbit from [here](../../../../test/e2e-v2/cases/redis/redis-exporter/fluent-bit.conf) for Redis.
 3. Config slow log from [here](../../../../test/e2e-v2/cases/redis/redis-exporter/redis.conf) for Redis.
 4. Periodically execute the [commands](../../../../test/e2e-v2/cases/redis/redis-exporter/scripts/slowlog.sh).

oap-server/server-library/library-client/pom.xml

@@ -38,12 +38,16 @@
             <artifactId>library-util</artifactId>
             <version>${project.version}</version>
         </dependency>
-
         <dependency>
             <groupId>org.apache.skywalking</groupId>
             <artifactId>library-elasticsearch-client</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>telemetry-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>io.grpc</groupId>

oap-server/server-library/library-client/src/main/java/org/apache/skywalking/oap/server/library/client/elasticsearch/ElasticSearchClient.java

@@ -34,11 +34,11 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import lombok.RequiredArgsConstructor;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.library.elasticsearch.requests.search.Query;
 import org.apache.skywalking.library.elasticsearch.response.Documents;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.library.elasticsearch.ElasticSearch;
 import org.apache.skywalking.library.elasticsearch.ElasticSearchBuilder;
@@ -55,12 +55,15 @@
 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.util.HealthChecker;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
 /**
  * ElasticSearchClient connects to the ES server by using ES client APIs.
  */
 @Slf4j
-@RequiredArgsConstructor
 public class ElasticSearchClient implements Client, HealthCheckable {
     public static final String TYPE = "type";
 
@@ -93,7 +96,14 @@ public class ElasticSearchClient implements Client, HealthCheckable {
 
     private final AtomicReference<ElasticSearch> es = new AtomicReference<>();
 
-    public ElasticSearchClient(String clusterNodes,
+    private final ModuleManager moduleManager;
+    private HistogramMetrics singleWriteHistogram;
+    private HistogramMetrics singleUpdateHistogram;
+    private HistogramMetrics singleDeleteHistogram;
+    private HistogramMetrics bulkWriteHistogram;
+
+    public ElasticSearchClient(ModuleManager moduleManager,
+                               String clusterNodes,
                                String protocol,
                                String trustStorePath,
                                String trustStorePass,
@@ -104,6 +114,7 @@ public ElasticSearchClient(String clusterNodes,
                                int socketTimeout,
                                int responseTimeout,
                                int numHttpClientThread) {
+        this.moduleManager = moduleManager;
         this.clusterNodes = clusterNodes;
         this.protocol = protocol;
         this.trustStorePath = trustStorePath;
@@ -119,6 +130,7 @@ public ElasticSearchClient(String clusterNodes,
 
     @Override
     public void connect() {
+        initTelemetry();
         final ElasticSearch oldOne = es.get();
 
         final ElasticSearchBuilder cb =
@@ -345,21 +357,27 @@ public SearchResponse searchIDs(String indexName, Iterable<String> ids) {
     }
 
     public void forceInsert(String indexName, String id, Map<String, Object> source) {
-        IndexRequestWrapper wrapper = prepareInsert(indexName, id, source);
-        Map<String, Object> params = ImmutableMap.of("refresh", "true");
-        es.get().documents().index(wrapper.getRequest(), params);
+        try (HistogramMetrics.Timer timer = singleWriteHistogram.createTimer()) {
+            IndexRequestWrapper wrapper = prepareInsert(indexName, id, source);
+            Map<String, Object> params = ImmutableMap.of("refresh", "true");
+            es.get().documents().index(wrapper.getRequest(), params);
+        }
     }
 
     public void forceUpdate(String indexName, String id, Map<String, Object> source) {
-        UpdateRequestWrapper wrapper = prepareUpdate(indexName, id, source);
-        Map<String, Object> params = ImmutableMap.of("refresh", "true");
-        es.get().documents().update(wrapper.getRequest(), params);
+        try (HistogramMetrics.Timer timer = singleUpdateHistogram.createTimer()) {
+            UpdateRequestWrapper wrapper = prepareUpdate(indexName, id, source);
+            Map<String, Object> params = ImmutableMap.of("refresh", "true");
+            es.get().documents().update(wrapper.getRequest(), params);
+        }
     }
 
     public void deleteById(String indexName, String id) {
-        indexName = indexNameConverter.apply(indexName);
-        Map<String, Object> params = ImmutableMap.of("refresh", "true");
-        es.get().documents().deleteById(indexName, TYPE, id, params);
+        try (HistogramMetrics.Timer timer = singleDeleteHistogram.createTimer()) {
+            indexName = indexNameConverter.apply(indexName);
+            Map<String, Object> params = ImmutableMap.of("refresh", "true");
+            es.get().documents().deleteById(indexName, TYPE, id, params);
+        }
     }
 
     public IndexRequestWrapper prepareInsert(String indexName, String id,
@@ -388,10 +406,53 @@ public BulkProcessor createBulkProcessor(int bulkActions,
                             .batchOfBytes(batchOfBytes)
                             .flushInterval(Duration.ofSeconds(flushInterval))
                             .concurrentRequests(concurrentRequests)
+                            .bulkMetrics(bulkWriteHistogram)
                             .build(es);
     }
 
     public String formatIndexName(String indexName) {
         return indexNameConverter.apply(indexName);
     }
+
+    private void initTelemetry() {
+        MetricsCreator metricsCreator = moduleManager.find(TelemetryModule.NAME)
+                                                     .provider()
+                                                     .getService(MetricsCreator.class);
+        if (singleWriteHistogram == null) {
+            singleWriteHistogram = metricsCreator.createHistogramMetric(
+                "elasticsearch_write_latency",
+                "Elasticsearch write/update/delete latency in seconds, bulk_write include write/update",
+                new MetricsTag.Keys("operation"),
+                new MetricsTag.Values("single_write"),
+                0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0
+            );
+        }
+        if (singleUpdateHistogram == null) {
+            singleUpdateHistogram = metricsCreator.createHistogramMetric(
+                "elasticsearch_write_latency",
+                "Elasticsearch write/update/delete latency in seconds, bulk_write include write/update",
+                new MetricsTag.Keys("operation"),
+                new MetricsTag.Values("single_update"),
+                0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0
+            );
+        }
+        if (singleDeleteHistogram == null) {
+            singleDeleteHistogram = metricsCreator.createHistogramMetric(
+                "elasticsearch_write_latency",
+                "Elasticsearch write/update/delete latency in seconds, bulk_write include write/update",
+                new MetricsTag.Keys("operation"),
+                new MetricsTag.Values("single_delete"),
+                0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0
+            );
+        }
+        if (bulkWriteHistogram == null) {
+            bulkWriteHistogram = metricsCreator.createHistogramMetric(
+                "elasticsearch_write_latency",
+                "Elasticsearch write/update/delete latency in seconds, bulk_write include write/update",
+                new MetricsTag.Keys("operation"),
+                new MetricsTag.Values("bulk_write"),
+                0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0
+            );
+        }
+    }
 }

oap-server/server-library/library-client/src/test/java/org/apache/skywalking/library/elasticsearch/bulk/ElasticSearchIT.java

@@ -31,10 +31,20 @@
 import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchScroller;
 import org.apache.skywalking.oap.server.library.client.elasticsearch.IndexRequestWrapper;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleProvider;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.none.MetricsCreatorNoop;
+import org.apache.skywalking.oap.server.telemetry.none.NoneTelemetryProvider;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
+import org.powermock.reflect.Whitebox;
 import org.testcontainers.elasticsearch.ElasticsearchContainer;
 import org.testcontainers.utility.DockerImageName;
 
@@ -46,8 +56,11 @@
 import java.util.UUID;
 import java.util.function.Function;
 
+import static org.mockito.Mockito.mock;
+
 @Slf4j
 public class ElasticSearchIT {
+    private ModuleManager moduleManager;
 
     public static Collection<Object[]> versions() {
         // noinspection resource
@@ -75,17 +88,33 @@ public static Collection<Object[]> versions() {
         });
     }
 
+    @BeforeEach
+    public void setUp() throws Exception {
+        moduleManager = mock(ModuleManager.class);
+        ModuleDefine storageModule = mock(ModuleDefine.class);
+        ModuleProvider provider = mock(ModuleProvider.class);
+        Mockito.when(provider.getModule()).thenReturn(storageModule);
+
+        NoneTelemetryProvider telemetryProvider = mock(NoneTelemetryProvider.class);
+        Mockito.when(telemetryProvider.getService(MetricsCreator.class))
+               .thenReturn(new MetricsCreatorNoop());
+        TelemetryModule telemetryModule = Mockito.spy(TelemetryModule.class);
+        Whitebox.setInternalState(telemetryModule, "loadedProvider", telemetryProvider);
+        Mockito.when(moduleManager.find(TelemetryModule.NAME)).thenReturn(telemetryModule);
+    }
+
     @ParameterizedTest(name = "version: {0}")
     @MethodSource("versions")
     public void indexOperate(final ElasticsearchContainer server,
                              final String namespace) {
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();
 
@@ -134,10 +163,11 @@ public void documentOperate(final ElasticsearchContainer server,
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();
 
@@ -209,10 +239,11 @@ public void templateOperate(final ElasticsearchContainer server,
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();
 
@@ -264,10 +295,11 @@ public void bulk(final ElasticsearchContainer server,
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();
 
@@ -297,10 +329,11 @@ public void bulkPer_1KB(final ElasticsearchContainer server,
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();
 
@@ -326,10 +359,11 @@ public void timeSeriesOperate(final ElasticsearchContainer server,
         server.start();
 
         final ElasticSearchClient client = new ElasticSearchClient(
-                server.getHttpHostAddress(),
-                "http", "", "", "test", "test",
-                indexNameConverter(namespace), 500, 6000,
-                0, 15
+            moduleManager,
+            server.getHttpHostAddress(),
+            "http", "", "", "test", "test",
+            indexNameConverter(namespace), 500, 6000,
+            0, 15
         );
         client.connect();