[Core][Metrics] Add Seatunnel Metrics module #2888

lvlv-feifei · 2022-09-25T07:35:39Z

Purpose of this pull request

Complete the code of seatuunel-metrics according to the framework of the issue, and add module seatuunel-metrics, including seatuunel-core, seatuunel-flink, seatuunel-spark three modules.
seatuunel-core is responsible for the basic definition of the seatunnel indicator system, including the definition of indicators and external interfaces
seatuunel-flink is responsible for the connection between seatunel and flink metrics, and accepts flink metrics
seatuunel-spark is responsible for the connection between seatuunel and spark metrics, and accepts spark metrics

Check list

Code changed are covered with tests, or it does not need tests for reason:
If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs

Hisoka-X · 2022-09-26T06:07:02Z

Please solved ci problems. Thanks

Hisoka-X · 2022-09-26T07:23:45Z

...nnel-apis/seatunnel-api-flink/src/main/java/org/apache/seatunnel/flink/FlinkEnvironment.java

+        if(!config.hasPath(ConfigKeyName.Metric_Class)){
+            return StreamExecutionEnvironment.getExecutionEnvironment();
+        }
+        //构建flink-metrics参数


Use english only

Hisoka-X · 2022-09-26T07:24:22Z

...nnel-apis/seatunnel-api-flink/src/main/java/org/apache/seatunnel/flink/FlinkEnvironment.java

@@ -322,4 +326,61 @@ private void setCheckpoint() {
        }
    }

+    public StreamExecutionEnvironment creatMetricStreamEEnvironment() {


Suggested change

public StreamExecutionEnvironment creatMetricStreamEEnvironment() {

public StreamExecutionEnvironment creatStreamEnvironment() {

Not only have metrics config will run this method.

Hisoka-X · 2022-09-26T07:26:29Z

...el-apis/seatunnel-api-flink/src/main/java/org/apache/seatunnel/flink/util/ConfigKeyName.java

@@ -45,4 +45,9 @@ private ConfigKeyName() {
    public static final String MIN_STATE_RETENTION_TIME = "execution.query.state.min-retention";
    public static final String STATE_BACKEND = "execution.state.backend";
    public static final String PLANNER = "execution.planner";
+    public static final String Metric_Interval = "execution.metrics.interval";


Use unified name in spark/flink. There config name is for SeaTunnel, not only for Flink or Spark.

Hisoka-X · 2022-09-26T07:28:03Z

...nnel-apis/seatunnel-api-flink/src/main/java/org/apache/seatunnel/flink/FlinkEnvironment.java

+                        .stringType()
+                        .noDefaultValue();
+
+        Configuration seatunnel_reporter = new Configuration().set(REPORTERS_LIST, "seatunnel_reporter").set(REPORTER_CLASS, "org.apache.seatunnel.metrics.flink.SeatunnelMetricReporter");


use lower camel case for field and parameter. Change seatunnel_reporter to seatunnelReporter

Hisoka-X · 2022-09-26T07:29:23Z

...nnel-apis/seatunnel-api-flink/src/main/java/org/apache/seatunnel/flink/FlinkEnvironment.java

+            return StreamExecutionEnvironment.getExecutionEnvironment();
+        }
+        //构建flink-metrics参数
+        ConfigOption<String> REPORTERS_LIST =


Maybe you can create an new method called initMetricConfig(Configuration config) to cover this logic.

Hisoka-X · 2022-09-26T07:34:19Z

seatunnel-metrics/seatunnel-metrics-core/pom.xml

+
+    <dependencies>
+        <dependency>
+            <groupId>io.prometheus</groupId>


Core can't contains prometheus, your should create new module call seatunnel-metrics-prometheus, then implement your metrics push to prometheus logic in there. The core should only have interface for metrics and common code.

Hisoka-X · 2022-09-26T07:35:36Z

.../src/main/java/org/apache/seatunnel/metrics/core/reporter/PrometheusPushGatewayReporter.java

+/**
+ * A reporter which outputs measurements to PrometheusPushGateway
+ */
+public class PrometheusPushGatewayReporter implements MetricReporter {


Move to seatunnel-metrics-prometheus

Hisoka-X · 2022-09-26T07:37:16Z

...trics/seatunnel-metrics-core/src/main/java/org/apache/seatunnel/metrics/core/MetricInfo.java

+        builder.append("helpString: ")
+                .append(this.helpString)
+                .append(lineSeparator);
+        for(int i=0;i<this.dimensionKeys.size();i++){


Code need be formated.

Hisoka-X · 2022-09-26T07:39:50Z

...-metrics-flink/src/main/java/org/apache/seatunnel/metrics/flink/SeatunnelMetricReporter.java

+        jobName = config.getString("jobName","flinkJob");
+        //config.
+        //String string = metricConfig.getString("name", "de");
+        //log.info("StreamMetricReporter init:{}", string);


Remove useless code.

Hisoka-X · 2022-09-26T07:40:14Z

...-metrics-flink/src/main/java/org/apache/seatunnel/metrics/flink/SeatunnelMetricReporter.java

+ * exports Flink metrics to Seatunnel
+ */
+public class SeatunnelMetricReporter extends AbstractSeatunnelReporter implements Scheduled {
+    private static final Logger log = LoggerFactory.getLogger(SeatunnelMetricReporter.class);


Suggested change

private static final Logger log = LoggerFactory.getLogger(SeatunnelMetricReporter.class);

private static final Logger LOGGER = LoggerFactory.getLogger(SeatunnelMetricReporter.class);

…sink (apache#2866) * parameter verification * update * update

* [Improve][DOC] Perfect the connector v2 doc * Update seatunnel-connectors-v2/README.zh.md Co-authored-by: Hisoka <fanjiaeminem@qq.com> * [Improve][DOC] A little tinkering * [Improve][DOC] A little tinkering * [Doc][connector] add Console sink doc close apache#2794 * [Doc][connector] add Console sink doc close apache#2794 * fix some problem * fix some problem * fine tuning Co-authored-by: Hisoka <fanjiaeminem@qq.com>

…itEnumerator (apache#2900)

…a types (apache#2860) * [Improve][Connector-V2] Improve orc write strategy to support all data types Co-authored-by: tyrantlucifer <tyrantlucifer@gmail.com>

…e#2878)

…(row) (apache#2839) * [Improve][Connector-v2]Supports direct definition of data values(row)

ashulin

Hey, I have a different idea about Metrics.
For various engines, they already have metrics and metric reporters, and engine users have also managed metrics based on their metrics reporters.
Seatunnel just submits an engine job, and it is unreasonable for users to do additional metrics management.
My opinion is that seatunnel provides unified metrics for the connector-v2 api, and then adds the metrics of seatunnel to the existing reporter of the engine.
For older versions of the connector, the engine's own metrics are used directly.
The metrics reporter of seatunnel may only be used for the engine of seatunnel itself.

Hisoka-X · 2022-09-27T03:42:26Z

My opinion is that seatunnel provides unified metrics for the connector-v2 api, and then adds the metrics of seatunnel to the existing reporter of the engine.

This is a feature will do next.

For older versions of the connector, the engine's own metrics are used directly.

There alway can use engine own metrics both old version connector and connector-v2

The metrics reporter of seatunnel may only be used for the engine of seatunnel itself.

At now, our reporter can support flink/spark/seatunnel with a little work. It is not conflict with their own reporter.

hailin0 · 2022-09-28T05:46:24Z

...-metrics-flink/src/main/java/org/apache/seatunnel/metrics/flink/SeatunnelMetricReporter.java

+            histogramsIndex.put(new SimpleHistogram(key.getCount(), key.getStatistics().getMin(), key.getStatistics().getMax(), key.getStatistics().getStdDev(), key.getStatistics().getMean(), quantile), metric.getValue());
+        }
+        //todo handle user config
+        reporter = new PrometheusPushGatewayReporter(jobName, host, port);


Moved outside the SeatunnelMetricReporter#report() method?

hailin0 · 2022-09-28T05:49:10Z

...eus/src/main/java/org/apache/seatunnel/metrics/prometheus/PrometheusPushGatewayReporter.java

+    @Override
+    public PrometheusPushGatewayReporter open() {
+        //todo Handle user config
+        return new PrometheusPushGatewayReporter("flink_prometheus_job", "localhost", DEFAULT_PORT);


Why create PrometheusPushGatewayReporter object again?

hailin0 · 2022-09-28T05:50:35Z

...el-metrics-core/src/main/java/org/apache/seatunnel/metrics/core/reporter/MetricReporter.java

+
+/** Reporters are used to export seatunnel {@link Metric Metrics} to an external backend. */
+public interface MetricReporter {
+    MetricReporter open();


Why return MetricReporter object? this is object method

hailin0 · 2022-09-28T05:51:37Z

...nel-metrics-spark/src/main/java/org/apache/seatunnel/metrics/spark/SeatunnelMetricSink.scala

+            }),
+          newMetricInfo(metricName, dimensionKeys, dimensionValues))
+      }
+      val reporter = new PrometheusPushGatewayReporter(pollJobName, pollHost, pollPort)


hailin0 · 2022-09-28T05:53:54Z

check file license header

# Conflicts: # seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-file-spark-e2e/src/test/resources/file/fakesource_to_local_orc.conf

2. Use reflection to automate assembly 3. Modify the flink/spark startup function 4. Try packaging configuration (todo)

Hisoka-X

LGTM

Hisoka-X · 2022-10-27T03:47:17Z

We merge it into st-metrics branch, should have long term to develop it before ready.

lvlv-feifei and others added 4 commits September 24, 2022 15:09

seatunnel-metrics init commit

97a80e5

Merge branch 'apache:dev' into dev

e9e7f20

seatunnel-metrics config add

4ace956

Merge branch 'apache:dev' into dev

94595c7

Hisoka-X changed the title ~~Seatunnel-metrics~~ [Core][Metrics] Add Seatunnel Metrics module Sep 26, 2022

Hisoka-X reviewed Sep 26, 2022

View reviewed changes

lvlv-feifei and others added 7 commits September 26, 2022 15:58

codeStyle update

e26fff0

codeStyle update again

7b96285

codeStyle seatunnel-spark update

5099f18

[Imporve][Connector-V2]Parameter verification for connector V2 kafka …

254223f

…sink (apache#2866) * parameter verification * update * update

add seatunnel-examples from gitignore (apache#2892)

12fa644

[Improve][connector-jdbc] Calculate splits only once in JdbcSourceSpl…

7622f28

…itEnumerator (apache#2900)

CalvinKirs added the feature New feature label Sep 26, 2022

hailin0 and others added 4 commits September 26, 2022 20:41

[Bug][Connector-V2] Fix wechat sink data serialization (apache#2856)

3aee11f

[Improve][Connector-V2] Improve orc write strategy to support all dat…

4d048cc

…a types (apache#2860) * [Improve][Connector-V2] Improve orc write strategy to support all data types Co-authored-by: tyrantlucifer <tyrantlucifer@gmail.com>

[Bug][seatunnel-translation-base] Fix Source restore state NPE (apach…

b6f7e77

…e#2878)

[Improve][Connector-v2-Fake]Supports direct definition of data values…

b7d9dde

…(row) (apache#2839) * [Improve][Connector-v2]Supports direct definition of data values(row)

ashulin marked this pull request as draft September 26, 2022 18:05

ashulin reviewed Sep 26, 2022

View reviewed changes

lvlv-feifei and others added 3 commits September 27, 2022 11:48

seatunnel-prometheus update

4b547aa

Merge remote-tracking branch 'origin/dev' into dev

9835c2a

Merge branch 'apache:dev' into dev

7de009e

hailin0 reviewed Sep 28, 2022

View reviewed changes

lvlv-feifei added 3 commits September 28, 2022 16:57

seatunnel-prometheus update

e9fc5ec

seatunnel-prometheus update

c178fb6

Merge remote-tracking branch 'origin/dev' into dev

af48276

# Conflicts: # seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-file-spark-e2e/src/test/resources/file/fakesource_to_local_orc.conf

1. Seatunnel unified configuration naming

7d7750e

2. Use reflection to automate assembly 3. Modify the flink/spark startup function 4. Try packaging configuration (todo)

Hisoka-X changed the base branch from dev to st-metrics October 20, 2022 02:17

Hisoka-X marked this pull request as ready for review October 20, 2022 02:19

Hisoka-X approved these changes Oct 27, 2022

View reviewed changes

Hisoka-X merged commit 0567e2a into apache:st-metrics Oct 27, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Core][Metrics] Add Seatunnel Metrics module #2888

[Core][Metrics] Add Seatunnel Metrics module #2888

lvlv-feifei commented Sep 25, 2022

Hisoka-X commented Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

Hisoka-X Sep 26, 2022

ashulin left a comment •

edited

Loading

Hisoka-X commented Sep 27, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 commented Sep 28, 2022

Hisoka-X left a comment

Hisoka-X commented Oct 27, 2022

	public StreamExecutionEnvironment creatMetricStreamEEnvironment() {
	public StreamExecutionEnvironment creatStreamEnvironment() {

	private static final Logger log = LoggerFactory.getLogger(SeatunnelMetricReporter.class);
	private static final Logger LOGGER = LoggerFactory.getLogger(SeatunnelMetricReporter.class);

[Core][Metrics] Add Seatunnel Metrics module #2888

[Core][Metrics] Add Seatunnel Metrics module #2888

Conversation

lvlv-feifei commented Sep 25, 2022

Purpose of this pull request

Check list

Hisoka-X commented Sep 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ashulin left a comment • edited Loading

Choose a reason for hiding this comment

Hisoka-X commented Sep 27, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hailin0 commented Sep 28, 2022

Hisoka-X left a comment

Choose a reason for hiding this comment

Hisoka-X commented Oct 27, 2022

ashulin left a comment •

edited

Loading