job execute collector

elasticjob-ecosystem/elasticjob-executor/elasticjob-executor-kernel/src/main/java/org/apache/shardingsphere/elasticjob/executor/ElasticJobExecutor.java

@@ -69,6 +69,9 @@ private ElasticJobExecutor(final ElasticJob elasticJob, final JobConfiguration j
         executorContext = new ExecutorContext(jobFacade.loadJobConfiguration(true));
         itemErrorMessages = new ConcurrentHashMap<>(jobConfig.getShardingTotalCount(), 1);
     }
+    public void init(String jobName){
+        jobFacade.postJobStatusTraceEvent("0", State.TASK_INIT, String.format("Job '%s' init success.", jobName));
+    }
 
     /**
      * Execute job.

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/event/JobStatusTraceEvent.java

@@ -57,7 +57,7 @@ public final class JobStatusTraceEvent implements JobEvent {
     private Date creationTime = new Date();
 
     public enum State {
-        TASK_STAGING, TASK_RUNNING, TASK_FINISHED, TASK_KILLED, TASK_LOST, TASK_FAILED, TASK_ERROR, TASK_DROPPED, TASK_GONE, TASK_GONE_BY_OPERATOR, TASK_UNREACHABLE, TASK_UNKNOWN
+        TASK_INIT, TASK_STAGING, TASK_RUNNING, TASK_FINISHED, TASK_KILLED, TASK_LOST, TASK_FAILED, TASK_ERROR, TASK_DROPPED, TASK_GONE, TASK_GONE_BY_OPERATOR, TASK_UNREACHABLE, TASK_UNKNOWN
     }
 
     public enum Source {

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-api/src/main/java/org/apache/shardingsphere/elasticjob/tracing/listener/TracingType.java

@@ -0,0 +1,8 @@
+package org.apache.shardingsphere.elasticjob.tracing.listener;
+
+public class TracingType {
+    public static final String RDB = "RDB";
+    public static final String METRICS = "METRICS";
+
+
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/pom.xml

@@ -0,0 +1,39 @@
+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.shardingsphere.elasticjob</groupId>
+        <artifactId>elasticjob-tracing</artifactId>
+        <version>3.1.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>elasticjob-tracing-observability</artifactId>
+    <packaging>jar</packaging>
+
+    <name>elasticjob-tracing-observability</name>
+    <url>http://maven.apache.org</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>3.8.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.shardingsphere.elasticjob</groupId>
+            <artifactId>elasticjob-tracing-api</artifactId>
+            <version>3.1.0-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-registry-prometheus</artifactId>
+            <version>1.11.3</version>
+        </dependency>
+    </dependencies>
+</project>

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/binder/JobMetrics.java

@@ -0,0 +1,22 @@
+package org.apache.shardingsphere.elasticjob.tracing.metrics.binder;
+
+import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class JobMetrics {
+    /**
+     * jobname -> state -> count
+     * support job level metrics
+     */
+    private Map<String, Map<JobStatusTraceEvent.State, AtomicLong>> metrics = new ConcurrentHashMap<>();
+
+    public void increase(String jobName, JobStatusTraceEvent.State state) {
+        metrics.computeIfAbsent(jobName, k -> new ConcurrentHashMap<>());
+        metrics.get(jobName).computeIfAbsent(state, k -> new AtomicLong(0)).incrementAndGet();
+    }
+
+
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/config/MetricConfig.java

@@ -0,0 +1,22 @@
+package org.apache.shardingsphere.elasticjob.tracing.metrics.config;
+
+public class MetricConfig {
+    private Integer metricsPort = 9090;
+    private String metricsPath = "/metrics";
+
+    public Integer getMetricsPort() {
+        return metricsPort;
+    }
+
+    public void setMetricsPort(Integer metricsPort) {
+        this.metricsPort = metricsPort;
+    }
+
+    public String getMetricsPath() {
+        return metricsPath;
+    }
+
+    public void setMetricsPath(String metricsPath) {
+        this.metricsPath = metricsPath;
+    }
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/config/MetricsTracingStorageConfiguration.java

@@ -0,0 +1,4 @@
+package org.apache.shardingsphere.elasticjob.tracing.metrics.config;
+
+public class MetricsTracingStorageConfiguration {
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/listener/JobMetricsListener.java

@@ -0,0 +1,92 @@
+/*
+ * 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.shardingsphere.elasticjob.tracing.metrics.listener;
+
+import com.sun.net.httpserver.HttpServer;
+import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
+import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent;
+import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListener;
+import org.apache.shardingsphere.elasticjob.tracing.metrics.binder.JobMetrics;
+import org.apache.shardingsphere.elasticjob.tracing.metrics.config.MetricConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+
+
+public class JobMetricsListener implements TracingListener {
+    private static final Logger logger = LoggerFactory.getLogger(JobMetricsListener.class);
+
+    private JobMetrics jobMetrics;
+
+    CompositeMeterRegistry composite = new CompositeMeterRegistry();
+
+
+    public JobMetricsListener(MetricConfig metricConfig) {
+        //open jvm http port to export metrics
+        jobMetrics = new JobMetrics();
+
+
+        //开启一个服务，用于暴露指标
+        HttpServer server = null;
+        try {
+            PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+            composite.add(prometheusRegistry);
+
+            server = HttpServer.create(new InetSocketAddress(8081), 0);
+
+            server.createContext("/metrics", httpExchange -> {
+                String response = prometheusRegistry.scrape();
+                httpExchange.sendResponseHeaders(200, response.getBytes().length);
+                try (OutputStream os = httpExchange.getResponseBody()) {
+                    os.write(response.getBytes());
+                }
+            });
+            //开启服务
+            new Thread(server::start).start();
+        } catch (IOException e) {
+            logger.error("metrics start error:{}", e.getMessage(), e);
+        }
+        HttpServer finalServer = server;
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            if (finalServer != null) {
+                finalServer.stop(3000);
+            }
+        }));
+
+    }
+
+    @Override
+    public void listen(JobExecutionEvent jobExecutionEvent) {
+
+    }
+
+    @Override
+    public void listen(JobStatusTraceEvent jobStatusTraceEvent) {
+        if (jobStatusTraceEvent == null || StringUtils.isBlank(jobStatusTraceEvent.getJobName())) {
+            return;
+        }
+        jobMetrics.increase(jobStatusTraceEvent.getJobName(), jobStatusTraceEvent.getState());
+    }
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/listener/JobMetricsTracingListenerConfiguration.java

@@ -0,0 +1,43 @@
+/*
+ * 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.shardingsphere.elasticjob.tracing.metrics.listener;
+
+import org.apache.shardingsphere.elasticjob.tracing.exception.TracingConfigurationException;
+import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListener;
+import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListenerConfiguration;
+import org.apache.shardingsphere.elasticjob.tracing.listener.TracingType;
+import org.apache.shardingsphere.elasticjob.tracing.metrics.config.MetricConfig;
+
+
+/**
+ * RDB tracing listener configuration.
+ */
+public final class JobMetricsTracingListenerConfiguration implements TracingListenerConfiguration<MetricConfig> {
+
+    @Override
+    public TracingListener createTracingListener(final MetricConfig metricConfig) throws TracingConfigurationException {
+        return new JobMetricsListener(metricConfig);
+    }
+
+    @Override
+    public String getType() {
+        return TracingType.METRICS;
+    }
+
+
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/java/org/apache/shardingsphere/elasticjob/tracing/metrics/package-info.java

@@ -0,0 +1 @@
+package org.apache.shardingsphere.elasticjob.tracing.metrics;

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.tracing.listener.TracingListenerConfiguration

@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+org.apache.shardingsphere.elasticjob.tracing.metrics.listener.JobMetricsTracingListenerConfiguration

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-observability/src/test/java/org/apache/shardingsphere/elasticjob/AppTest.java

@@ -0,0 +1,38 @@
+package org.apache.shardingsphere.elasticjob;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+/**
+ * Unit test for simple App.
+ */
+public class AppTest 
+    extends TestCase
+{
+    /**
+     * Create the test case
+     *
+     * @param testName name of the test case
+     */
+    public AppTest( String testName )
+    {
+        super( testName );
+    }
+
+    /**
+     * @return the suite of tests being tested
+     */
+    public static Test suite()
+    {
+        return new TestSuite( AppTest.class );
+    }
+
+    /**
+     * Rigourous Test :-)
+     */
+    public void testApp()
+    {
+        assertTrue( true );
+    }
+}

elasticjob-ecosystem/elasticjob-tracing/elasticjob-tracing-rdb/src/main/java/org/apache/shardingsphere/elasticjob/tracing/rdb/listener/RDBTracingListenerConfiguration.java

@@ -20,6 +20,7 @@
 import org.apache.shardingsphere.elasticjob.tracing.exception.TracingConfigurationException;
 import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListener;
 import org.apache.shardingsphere.elasticjob.tracing.listener.TracingListenerConfiguration;
+import org.apache.shardingsphere.elasticjob.tracing.listener.TracingType;
 
 import javax.sql.DataSource;
 import java.sql.SQLException;
@@ -40,6 +41,6 @@ public TracingListener createTracingListener(final DataSource storage) throws Tr
 
     @Override
     public String getType() {
-        return "RDB";
+        return TracingType.RDB;
     }
 }

elasticjob-ecosystem/elasticjob-tracing/pom.xml

@@ -30,5 +30,6 @@
     <modules>
         <module>elasticjob-tracing-api</module>
         <module>elasticjob-tracing-rdb</module>
+        <module>elasticjob-tracing-observability</module>
     </modules>
 </project>