first commit

hippo4j-example/pom.xml

@@ -13,6 +13,7 @@
     <modules>
         <module>hippo4j-example-core</module>
         <module>hippo4j-spring-boot-starter-example</module>
+        <module>hippo4j-spring-boot-starter-es-monitor-example</module>
         <module>hippo4j-core-nacos-spring-boot-starter-example</module>
         <module>hippo4j-core-apollo-spring-boot-starter-example</module>
         <module>hippo4j-core-zookeeper-spring-boot-starter-example</module>

hippo4j-monitor/hippo4j-monitor-base/src/main/java/com/example/monitor/base/MonitorTypeEnum.java

@@ -22,5 +22,5 @@ package com.example.monitor.base;
  */
 public enum MonitorTypeEnum {
 
-    LOG, PROMETHEUS, SERVER
+    LOG, PROMETHEUS, SERVER, ES
 }

hippo4j-monitor/hippo4j-monitor-es/pom.xml

@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>cn.hippo4j</groupId>
+        <artifactId>hippo4j-monitor</artifactId>
+        <version>${revision}</version>
+    </parent>
+    <artifactId>hippo4j-monitor-es</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>cn.hippo4j</groupId>
+            <artifactId>hippo4j-monitor-base</artifactId>
+            <version>${revision}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-core</artifactId>
+            <version>${micrometer-core.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>${elasticsearch.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>transport</artifactId>
+            <version>${elasticsearch.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>${elasticsearch.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <includes>
+                    <include>**/*.txt</include>
+                    <include>**/*.json</include>
+                </includes>
+            </resource>
+        </resources>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <configuration>
+                    <archive>
+                        <manifestEntries>
+                            <Implementation-Title>${project.artifactId}</Implementation-Title>
+                            <Implementation-Version>${project.version}</Implementation-Version>
+                            <Build-Time>${maven.build.timestamp}</Build-Time>
+                            <Built-By>chen.ma</Built-By>
+                        </manifestEntries>
+                    </archive>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-javadoc-plugin</artifactId>
+                <version>2.10.3</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

hippo4j-monitor/hippo4j-monitor-es/src/main/java/cn/hippo4j/monitor/es/EsMonitorHandler.java

@@ -0,0 +1,76 @@
+/*
+ * 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 cn.hippo4j.monitor.es;
+
+import cn.hippo4j.common.config.ApplicationContextHolder;
+import cn.hippo4j.common.model.ThreadPoolRunStateInfo;
+import cn.hippo4j.core.executor.state.ThreadPoolRunStateHandler;
+import cn.hutool.core.bean.BeanUtil;
+import com.example.monitor.base.AbstractDynamicThreadPoolMonitor;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.springframework.core.env.Environment;
+
+import java.util.Map;
+
+/**
+ * Prometheus monitor handler.
+ */
+public class EsMonitorHandler extends AbstractDynamicThreadPoolMonitor {
+
+    private final Map<String, ThreadPoolRunStateInfo> RUN_STATE_CACHE = Maps.newConcurrentMap();
+
+    public EsMonitorHandler(ThreadPoolRunStateHandler threadPoolRunStateHandler) {
+        super(threadPoolRunStateHandler);
+    }
+
+    @Override
+    protected void execute(ThreadPoolRunStateInfo poolRunStateInfo) {
+        Environment environment = ApplicationContextHolder.getInstance().getEnvironment();
+        RestHighLevelClient esClient = ApplicationContextHolder.getBean(RestHighLevelClient.class);
+
+        String applicationName = environment.getProperty("spring.application.name", "application");
+        // Iterable<Tag> tags = Lists.newArrayList(
+        // Tag.of(DYNAMIC_THREAD_POOL_ID_TAG, poolRunStateInfo.getTpId()),
+        // Tag.of(APPLICATION_NAME_TAG, applicationName));
+        // // load
+        // Metrics.gauge(metricName("current.load"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getSimpleCurrentLoad);
+        // Metrics.gauge(metricName("peak.load"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getSimplePeakLoad);
+        // // thread pool
+        // Metrics.gauge(metricName("core.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getCoreSize);
+        // Metrics.gauge(metricName("maximum.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getMaximumSize);
+        // Metrics.gauge(metricName("current.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getPoolSize);
+        // Metrics.gauge(metricName("largest.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getLargestPoolSize);
+        // Metrics.gauge(metricName("active.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getActiveSize);
+        // // queue
+        // Metrics.gauge(metricName("queue.size"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getQueueSize);
+        // Metrics.gauge(metricName("queue.capacity"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getQueueCapacity);
+        // Metrics.gauge(metricName("queue.remaining.capacity"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getQueueRemainingCapacity);
+        // // other
+        // Metrics.gauge(metricName("completed.task.count"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getCompletedTaskCount);
+        // Metrics.gauge(metricName("reject.count"), tags, poolRunStateInfo, ThreadPoolRunStateInfo::getRejectCount);
+    }
+
+    @Override
+    public String getType() {
+        return "es";
+    }
+}

hippo4j-monitor/hippo4j-monitor-es/src/main/java/cn/hippo4j/monitor/es/config/EsConfig.java

@@ -0,0 +1,148 @@
+/*
+ * 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 cn.hippo4j.monitor.es.config;
+
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+@Slf4j
+@Configuration
+@ConfigurationProperties(prefix = "es")
+@Getter
+@Setter
+public class EsConfig {
+
+    // cluster name
+    private String cluster;
+    // such as ip1:9200,ip2:9200
+    private String host;
+    // such as :http
+    private String scheme;
+    private String userName;
+    private String password;
+    private int timeout;
+
+    private RestHighLevelClient client;
+    @Getter
+    private SearchSourceBuilder searchBuilder;
+
+    public EsConfig() {
+    }
+
+    public EsConfig(String host) {
+        initRestClient(host, null, null);
+    }
+
+    public EsConfig(String host, String userName, String password) {
+        initRestClient(host, userName, password);
+    }
+
+    public RestHighLevelClient initRestClient(String host, String userName, String password) {
+        this.host = host;
+        this.userName = userName;
+        this.password = password;
+
+        return initRestClient();
+    }
+
+    @Bean(destroyMethod = "close")
+    protected RestHighLevelClient initRestClient() {
+        try {
+            List<HttpHost> hosts = parseHosts();
+
+            if (Strings.isNullOrEmpty(userName) || Strings.isNullOrEmpty(password)) {
+                client = new RestHighLevelClient(RestClient.builder(hosts.toArray(new HttpHost[]{})));
+            } else {
+                client = new RestHighLevelClient(RestClient.builder(hosts.toArray(new HttpHost[]{}))
+                        .setHttpClientConfigCallback(httpAsyncClientBuilder -> httpAsyncClientBuilder.setDefaultCredentialsProvider(getCredentialsProvider())));
+            }
+
+            log.info("[ES RestHighLevelClient] ES连接成功！host:{},scheme:{}", host, scheme);
+            return client;
+        } catch (Exception ex) {
+            log.error("[ES RestHighLevelClient] ES连接失败! cause:{}", Throwables.getStackTraceAsString(ex));
+        }
+        return null;
+    }
+
+    @Bean
+    protected BasicCredentialsProvider getCredentialsProvider() {
+        if (!Strings.isNullOrEmpty(userName) && !Strings.isNullOrEmpty(password)) {
+            final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+            credentialsProvider.setCredentials(AuthScope.ANY,
+                    new UsernamePasswordCredentials(userName, password));
+            return credentialsProvider;
+        }
+        return null;
+    }
+
+    public RestHighLevelClient getClient() {
+        return null == client ? initRestClient() : client;
+    }
+
+    @Bean
+    public SearchSourceBuilder getSearchSourceBuilder() {
+        if (null == searchBuilder) {
+            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
+            sourceBuilder.timeout(new TimeValue(timeout));
+
+            searchBuilder = sourceBuilder;
+        }
+
+        return searchBuilder;
+    }
+
+    private List<HttpHost> parseHosts() {
+        String[] hostAndPorts = host.split(",");
+
+        List<HttpHost> hosts = Lists.newArrayList();
+        for (String hostAndPort : hostAndPorts) {
+            hosts.add(new HttpHost(hostAndPort.split(":")[0], Integer.parseInt(hostAndPort.split(":")[1]), scheme));
+        }
+        return hosts;
+    }
+
+    public void printDebugConfigs() {
+        String debug = "EsSettings: \n" +
+                "cluster:" + cluster + "\n" +
+                "host:" + host + "\n" +
+                "scheme:" + scheme + "\n" +
+                "userName:" + userName + "\n" +
+                "password:" + password + "\n" +
+                "timeout:" + timeout;
+
+        log.info(debug);
+    }
+}

hippo4j-monitor/pom.xml

@@ -14,6 +14,7 @@
         <module>hippo4j-monitor-base</module>
         <module>hippo4j-monitor-log</module>
         <module>hippo4j-monitor-prometheus</module>
+        <module>hippo4j-monitor-es</module>
     </modules>
 
     <build>

hippo4j-spring-boot/hippo4j-core-spring-boot-starter/pom.xml

@@ -83,6 +83,12 @@
             <artifactId>hippo4j-monitor-prometheus</artifactId>
             <version>${revision}</version>
         </dependency>
+
+        <dependency>
+            <groupId>cn.hippo4j</groupId>
+            <artifactId>hippo4j-monitor-es</artifactId>
+            <version>${revision}</version>
+        </dependency>
     </dependencies>
 
     <build>

hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/config/MonitorHandlerConfiguration.java

@@ -18,8 +18,10 @@
 package cn.hippo4j.core.springboot.starter.config;
 
 import cn.hippo4j.core.executor.state.ThreadPoolRunStateHandler;
+import cn.hippo4j.core.springboot.starter.config.condition.EsMonitorCondition;
 import cn.hippo4j.core.springboot.starter.config.condition.LogMonitorCondition;
 import cn.hippo4j.core.springboot.starter.config.condition.PrometheusMonitorCondition;
+import cn.hippo4j.monitor.es.EsMonitorHandler;
 import cn.hippo4j.monitor.log.LogMonitorHandler;
 import cn.hippo4j.monitor.prometheus.PrometheusMonitorHandler;
 import org.springframework.context.annotation.Bean;
@@ -49,4 +51,13 @@ public class MonitorHandlerConfiguration {
             return new PrometheusMonitorHandler(threadPoolRunStateHandler);
         }
     }
+
+    @Conditional(EsMonitorCondition.class)
+    static class EmbeddedEsMonitor {
+
+        @Bean
+        public EsMonitorHandler EsMonitorHandler(ThreadPoolRunStateHandler threadPoolRunStateHandler) {
+            return new EsMonitorHandler(threadPoolRunStateHandler);
+        }
+    }
 }

hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/config/condition/EsMonitorCondition.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 cn.hippo4j.core.springboot.starter.config.condition;
+
+import cn.hippo4j.common.toolkit.StringUtil;
+import cn.hippo4j.core.springboot.starter.config.BootstrapCoreProperties;
+import com.example.monitor.base.MonitorTypeEnum;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * Prometheus monitor condition.
+ */
+public class EsMonitorCondition implements Condition {
+
+    @Override
+    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+        String collectType = context.getEnvironment().getProperty(BootstrapCoreProperties.PREFIX + "collect-type", "");
+        return StringUtil.isNotEmpty(collectType) && collectType.contains(MonitorTypeEnum.ES.name().toLowerCase()) ? true : false;
+    }
+}

hippo4j-spring-boot/hippo4j-spring-boot-starter/pom.xml

@@ -102,6 +102,12 @@
             <artifactId>hippo4j-monitor-prometheus</artifactId>
             <version>${revision}</version>
         </dependency>
+
+        <dependency>
+            <groupId>cn.hippo4j</groupId>
+            <artifactId>hippo4j-monitor-es</artifactId>
+            <version>${revision}</version>
+        </dependency>
     </dependencies>
 
     <build>

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/monitor/ReportingEventExecutor.java

@@ -112,7 +112,8 @@ public class ReportingEventExecutor implements Runnable, CommandLineRunner, Disp
                 new Integer(collectType.split(",").length),
                 ThreadFactoryBuilder.builder().daemon(true).prefix("client.scheduled.collect.data").build());
         if (collectType.contains(MonitorTypeEnum.PROMETHEUS.name().toLowerCase())
-                || collectType.contains(MonitorTypeEnum.LOG.name().toLowerCase())) {
+                || collectType.contains(MonitorTypeEnum.LOG.name().toLowerCase())
+                || collectType.contains(MonitorTypeEnum.ES.name().toLowerCase())) {
             // Get all dynamic thread pool monitoring components.
             threadPoolMonitors = ApplicationContextHolder.getBeansOfType(ThreadPoolMonitor.class);
             collectVesselExecutor.scheduleWithFixedDelay(

pom.xml

@@ -57,6 +57,7 @@
         <spring-cloud-starter-stream-rocketmq.version>2.2.6.RELEASE</spring-cloud-starter-stream-rocketmq.version>
         <spring-cloud-starter-netflix-hystrix.version>2.2.9.RELEASE</spring-cloud-starter-netflix-hystrix.version>
         <spring-cloud-starter-alibaba-nacos-config.version>2.2.5.RELEASE</spring-cloud-starter-alibaba-nacos-config.version>
+        <elasticsearch.version>6.5.0</elasticsearch.version>
 
         <!-- Maven & Build -->
         <java.version>1.8</java.version>