Copilot commented on code in PR #60493:
URL: https://github.com/apache/doris/pull/60493#discussion_r2762869007
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -633,6 +646,11 @@ private void
updateNoTxnJobStatisticAndOffset(CommitOffsetRequest offsetRequest)
.setScannedRows(this.nonTxnJobStatistic.getScannedRows() +
offsetRequest.getScannedRows());
this.nonTxnJobStatistic.setLoadBytes(this.nonTxnJobStatistic.getLoadBytes() +
offsetRequest.getScannedBytes());
offsetProvider.updateOffset(offsetProvider.deserializeOffset(offsetRequest.getOffset()));
+
+ //update metric
+
MetricRepo.COUNTER_STREAMING_JOB_TOTAL_ROWS.increase(offsetRequest.getScannedRows());
+
MetricRepo.COUNTER_STREAMING_JOB_FILTER_ROWS.increase(offsetRequest.getFilteredRows());
+
MetricRepo.COUNTER_STREAMING_JOB_LOAD_BYTES.increase(offsetRequest.getLoadBytes());
Review Comment:
`CommitOffsetRequest` (imported from
`org.apache.doris.httpv2.rest.StreamingJobAction`) only defines `jobId`,
`taskId`, `offset`, `scannedRows`, and `scannedBytes`. The calls to
`offsetRequest.getFilteredRows()` and `offsetRequest.getLoadBytes()` do not
exist on that type and will not compile. Either extend `CommitOffsetRequest` to
include these fields (and ensure the REST endpoint + sender populate them), or
change the metric updates to use available fields (e.g., `scannedBytes`) and
drop/replace the unavailable ones.
```suggestion
MetricRepo.COUNTER_STREAMING_JOB_FILTER_ROWS.increase(0L);
MetricRepo.COUNTER_STREAMING_JOB_LOAD_BYTES.increase(offsetRequest.getScannedBytes());
```
##########
fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java:
##########
@@ -638,6 +652,35 @@ public Long getValue() {
MetricUnit.NOUNIT, "task execute count of routine load");
DORIS_METRIC_REGISTER.addMetrics(COUNTER_ROUTINE_LOAD_TASK_EXECUTE_COUNT);
+ // streaming job metrics
+ COUNTER_STREAMING_JOB_GET_META_LANTENCY = new
LongCounterMetric("streaming_job_get_meta_latency",
+ MetricUnit.MILLISECONDS, "get meta lantency of streaming job");
+
DORIS_METRIC_REGISTER.addMetrics(COUNTER_STREAMING_JOB_GET_META_LANTENCY);
Review Comment:
Spelling: `LANTENCY`/`lantency` should be `LATENCY`/`latency` in the counter
constant name and help text. This is newly introduced for streaming job metrics
and makes the API harder to search/maintain.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -633,6 +646,11 @@ private void
updateNoTxnJobStatisticAndOffset(CommitOffsetRequest offsetRequest)
.setScannedRows(this.nonTxnJobStatistic.getScannedRows() +
offsetRequest.getScannedRows());
this.nonTxnJobStatistic.setLoadBytes(this.nonTxnJobStatistic.getLoadBytes() +
offsetRequest.getScannedBytes());
offsetProvider.updateOffset(offsetProvider.deserializeOffset(offsetRequest.getOffset()));
+
+ //update metric
+
MetricRepo.COUNTER_STREAMING_JOB_TOTAL_ROWS.increase(offsetRequest.getScannedRows());
+
MetricRepo.COUNTER_STREAMING_JOB_FILTER_ROWS.increase(offsetRequest.getFilteredRows());
+
MetricRepo.COUNTER_STREAMING_JOB_LOAD_BYTES.increase(offsetRequest.getLoadBytes());
Review Comment:
These counters (`streaming_job_total_rows/filter_rows/load_bytes`) are only
increased in the non-txn `commitOffset()` path
(`updateNoTxnJobStatisticAndOffset`). The txn-based streaming insert path
updates stats via `StreamingTaskTxnCommitAttachment` in
`afterCommitted()/replayOnCommitted()`, but does not update the same metrics,
so the exported totals will undercount for streaming insert jobs that run via
txn attachments. Consider updating the same counters in the txn-based statistic
update paths as well (or clarify via naming/labels that the metrics are only
for the non-txn multi-table mode).
##########
regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_metrics.groovy:
##########
@@ -0,0 +1,159 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_mysql_job_metrics",
+ "p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
+
+ def jobName = "test_streaming_mysql_job_metrics"
+ def currentDb = (sql "select database()")[0][0]
+ def mysqlDb = "test_cdc_db"
+ def mysqlTable = "user_info_metrics"
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysql_port = context.config.otherConfigs.get("mysql_57_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3_endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driver_url =
"https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
+ sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
+ sql """DROP TABLE IF EXISTS ${mysqlDb}.${mysqlTable}"""
+ sql """CREATE TABLE ${mysqlDb}.${mysqlTable} (
+ `name` varchar(200) NOT NULL,
+ `age` int DEFAULT NULL,
+ PRIMARY KEY (`name`)
+ ) ENGINE=InnoDB"""
+ sql """INSERT INTO ${mysqlDb}.${mysqlTable} (name, age) VALUES
('Alice', 10)"""
+ sql """INSERT INTO ${mysqlDb}.${mysqlTable} (name, age) VALUES
('Bob', 20)"""
+ }
+
+ // create streaming job: FROM MYSQL ... TO DATABASE currentDb
+ sql """
+ CREATE JOB ${jobName}
+ ON STREAMING
+ FROM MYSQL (
+ "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
+ "driver_url" = "${driver_url}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root",
+ "password" = "123456",
+ "database" = "${mysqlDb}",
+ "include_tables" = "${mysqlTable}",
+ "offset" = "initial"
+ )
+ TO DATABASE ${currentDb} (
+ "table.create.properties.replication_num" = "1"
+ )
+ """
+
+ try {
+ Awaitility.await().atMost(300, SECONDS)
+ .pollInterval(1, SECONDS).until({
+ def jobInfo = sql """
+ select SucceedTaskCount, Status
+ from jobs("type"="insert")
+ where Name = '${jobName}' and
ExecuteType='STREAMING'
+ """
+ log.info("metrics job status: " + jobInfo)
+ jobInfo.size() == 1 &&
+ Integer.parseInt(jobInfo[0][0] as String) >= 1
&&
+ (jobInfo[0][1] as String) == "RUNNING"
+ })
+ } catch (Exception ex) {
+ def showjob = sql """select * from jobs("type"="insert") where
Name='${jobName}'"""
+ def showtask = sql """select * from tasks("type"="insert") where
JobName='${jobName}'"""
+ log.info("metrics show job: " + showjob)
+ log.info("metrics show task: " + showtask)
+ throw ex
+ }
+
+ int count = 0
+ int metricCount = 0
+ while (true) {
+ metricCount = 0
+ httpTest {
+ endpoint context.config.feHttpAddress
+ uri "/metrics?type=json"
+ op "get"
+ check { code, body ->
+ logger.debug("code:${code} body:${body}")
+
+ if
(body.contains("doris_fe_streaming_job_get_meta_latency")) {
+ log.info("contain
doris_fe_streaming_job_get_meta_latency")
+ metricCount++
+ }
+ if
(body.contains("doris_fe_streaming_job_get_meta_count")) {
+ log.info("contain
doris_fe_streaming_job_get_meta_count")
+ metricCount++
+ }
+ if
(body.contains("doris_fe_streaming_job_get_meta_fail_count")) {
+ log.info("contain
doris_fe_streaming_job_get_meta_fail_count")
+ metricCount++
+ }
+ if
(body.contains("doris_fe_streaming_job_task_execute_time")) {
+ log.info("contain
doris_fe_streaming_job_task_execute_time")
+ metricCount++
+ }
+ if
(body.contains("doris_fe_streaming_job_task_execute_count")) {
+ log.info("contain
doris_fe_streaming_job_task_execute_count")
+ metricCount++
+ }
+ if
(body.contains("doris_fe_streaming_job_task_failed_count")) {
+ log.info("contain
doris_fe_streaming_job_task_failed_count")
+ metricCount++
+ }
+ if (body.contains("doris_fe_streaming_job_total_rows")) {
+ log.info("contain doris_fe_streaming_job_total_rows")
+ metricCount++
+ }
+ if (body.contains("doris_fe_streaming_job_filter_rows")) {
+ log.info("contain doris_fe_streaming_job_filter_rows")
+ metricCount++
+ }
+ if (body.contains("doris_fe_streaming_job_load_bytes")) {
+ log.info("contain doris_fe_streaming_job_load_bytes")
+ metricCount++
+ }
+ }
+ }
+
+ if (metricCount >= 9) {
+ break
+ }
+
+ count++
+ sleep(1000)
+ if (count > 60) {
+ // timeout, failed
+ assertEquals(1, 2)
+ }
+ }
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ def jobCountRsp = sql """select count(1) from jobs("type"="insert")
where Name ='${jobName}'"""
+ assert jobCountRsp.get(0).get(0) == 0
Review Comment:
This test creates an external MySQL database/table and a streaming job, but
cleanup (`DROP JOB ...`) is only executed on the success path. If the
Awaitility wait or the metrics polling loop times out/throws, the job can be
left running and affect subsequent suites. Wrap job creation/verification in
`try/finally` to always drop the job (and ideally drop the MySQL table/database
as well).
##########
fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java:
##########
@@ -1060,6 +1103,49 @@ public Long getValue() {
DORIS_METRIC_REGISTER.addMetrics(gauge);
}
+ private static void initStreamingJobMetrics() {
+ // streaming insert jobs
+ for (JobStatus jobStatus : JobStatus.values()) {
+ if (jobStatus == JobStatus.PAUSED) {
+ addStreamingJobStateGaugeMetric(jobStatus, "USER_PAUSED",
+ job -> job.getFailureReason() != null
+ && job.getFailureReason().getCode() ==
InternalErrorCode.MANUAL_PAUSE_ERR);
+ addStreamingJobStateGaugeMetric(jobStatus, "ABNORMAL_PAUSED",
+ job -> job.getFailureReason() != null
+ && job.getFailureReason().getCode() !=
InternalErrorCode.MANUAL_PAUSE_ERR);
+ }
+ addStreamingJobStateGaugeMetric(jobStatus, jobStatus.name(), job
-> true);
+ }
+ }
+
+ private static void addStreamingJobStateGaugeMetric(
+ JobStatus jobStatus, String stateLabel,
Predicate<StreamingInsertJob> filter) {
+
+ GaugeMetric<Long> gauge = new GaugeMetric<Long>(
+ "job", MetricUnit.NOUNIT, "streaming job statistics") {
+ @Override
+ public Long getValue() {
+ if (!Env.getCurrentEnv().isMaster()) {
+ return 0L;
+ }
+ List<org.apache.doris.job.base.AbstractJob> jobs =
+
Env.getCurrentEnv().getJobManager().queryJobs(org.apache.doris.job.common.JobType.INSERT);
+
+ return jobs.stream()
+ .filter(job -> job instanceof StreamingInsertJob)
+ .map(job -> (StreamingInsertJob) job)
+ .filter(job -> job.getJobStatus() == jobStatus)
+ .filter(filter)
+ .count();
+ }
+ };
+
+ gauge.addLabel(new MetricLabel("job", "load"))
+ .addLabel(new MetricLabel("type", "STREAMING_JOB"))
+ .addLabel(new MetricLabel("state", stateLabel));
+ DORIS_METRIC_REGISTER.addMetrics(gauge);
Review Comment:
The new streaming-job state gauges reuse the existing metric name `job` and
the same `state` label values (e.g., `USER_PAUSED` / `ABNORMAL_PAUSED`) as
routine load. Because `DorisMetricRegistry.MetricList` stores metrics in a
`HashMap` and iteration order is non-deterministic, consumers that select only
by `metric`+`state` can become flaky/incorrect. This is likely to break the
existing regression test
`regression-test/suites/load_p0/routine_load/test_routin_load_abnormal_job_monitor.groovy`
(it does `find { tags.metric=="doris_fe_job" && tags.state=="ABNORMAL_PAUSED"
}` without filtering `type`). Suggested fix: update that test to also filter by
`type=="ROUTINE_LOAD"`, or change the streaming-job state metric naming/labels
(e.g., distinct metric name instead of `job`) to avoid collisions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
