Repository: lf-edge/ekuiper
Pull request #3188 (2024-09-11 07:29:35)

feat(file): support decorator

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `23.07692%` with `30 lines` in your changes missing coverage. Please review.
> Project coverage is 71.08%. Comparing base [(`09c3488`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/09c3488739e355b4038a169df5111e2474752c9b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) to head [(`279633c`)](https://app.codecov.io/gh/lf-edge/ekuiper/commit/279633c9bf7c5849b804bc914390d4a3d658f56a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).

| [Files with missing lines](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/file/source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&filepath=internal%2Fio%2Ffile%2Fsource.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vZmlsZS9zb3VyY2UuZ28=) | 20.00% | [12 Missing and 4 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [pkg/cast/time.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&filepath=pkg%2Fcast%2Ftime.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL2Nhc3QvdGltZS5nbw==) | 30.00% | [7 Missing :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [pkg/modules/file.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&filepath=pkg%2Fmodules%2Ffile.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-cGtnL21vZHVsZXMvZmlsZS5nbw==) | 22.22% | [6 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3188      +/-   ##
==========================================
- Coverage   71.12%   71.08%   -0.05%     
==========================================
  Files         386      386              
  Lines       43601    43637      +36     
==========================================
+ Hits        31011    31015       +4     
- Misses      10137    10164      +27     
- Partials     2453     2458       +5     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/3188?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/docs/directory.json b/docs/directory.json
index 94e4bf23..468a8bce 100644
--- a/docs/directory.json
+++ b/docs/directory.json
@@ -348,6 +348,10 @@
               ]
             }
           ]
+        },
+        {
+          "title": "连接管理",
+          "path": "guide/connections/overview"
         }
       ]
     },
@@ -1120,6 +1124,10 @@
               ]
             }
           ]
+        },
+        {
+          "title": "Connection Management",
+          "path": "guide/connections/overview"
         }
       ]
     },
diff --git a/docs/en_US/guide/connections/overview.md b/docs/en_US/guide/connections/overview.md
new file mode 100644
index 00000000..fe22a29b
--- /dev/null
+++ b/docs/en_US/guide/connections/overview.md
@@ -0,0 +1,134 @@
+# Connection Management
+
+Sources and Sinks are used for interacting with external systems, which involve connecting to external resources. This
+chapter primarily explains how eKuiper manages connections.
+
+## Connection Types
+
+Different connection types vary in complexity. For example, MQTT long connections require attention to connection
+status, and there may be disconnections during rule execution that necessitate automatic reconnection. On the other
+hand, HTTP connections are stateless by default, making their state management simpler. To unify the management of
+complex connection resources, including creation, reuse, automatic reconnection, and obtaining connection status,
+eKuiper v2 introduced an internal connection pool component and adapted a series of connection types:
+
+- MQTT Connection
+- Neuron Connection
+- EdgeX Connection
+- SQL Connection
+- HTTP Connection (including REST sink, HTTP Pull source, and HTTP push source connections)
+- WebSocket Connection
+
+Other connection types may be gradually integrated in subsequent versions. Connection types integrated into the
+connection pool can be independently created via API and accessed.
+
+In eKuiper, the lifecycle management of various connections is divided into three types:
+
+1. **Connection Attached to Rule**: By default, connections are managed by the Source/Sink implementation itself, with
+   their lifecycle controlled by the rule in use. The connection resources will only start connecting when the rule is
+   started; when the rule ends, the connection will be closed. In the example below, we create a data stream `memStream`
+   of type `memory`. Since this type is not integrated into the connection pool, the connection will only be established
+   when a rule using this stream is started.
+
+    ```sql
+    create stream memStream () WITH (TYPE="mqtt", DATASOURCE="demo")
+    ```
+
+2. **Anonymous Connection Resource Managed by Connection Pool**: Some connection types are adapted to the connection
+   pool management interface, with their lifecycle managed by the connection pool. When a rule containing these types of
+   connections is started, the rule will obtain an anonymous (actual resource id generated by the rule and not shared)
+   resource from the connection pool. In the example below, we create a data stream `mqttStream` of type `mqtt`. The
+   connection is anonymous, and since this type is adapted to the connection pool, we can retrieve this connection via
+   the connection API. When the rule is deleted, the corresponding connection will also be deleted.
+
+    ```sql
+    create stream mqttStream () WITH (TYPE="mqtt", DATASOURCE="demo")
+    ```
+
+3. **User-Created Connection Resource**: Users can add, delete, modify, and query resources via
+   the [Connection Management API](../../api/restapi/connection.md). Resources created via the API must specify a unique
+   id, which can be referenced in rules. **Note**: Only connection resources adapted to the connection pool can be
+   managed via the API. Connections created in this way are independent physical connections that run immediately after
+   creation and do not depend on rules. They can be shared by multiple rules or multiple sources/sinks.
+
+**Note**: User-created connections are physical connections that will automatically reconnect until the connection is
+successful.
+
+### Connection Reuse
+
+User-created connection resources can run independently, and multiple rules can reference this named resource.
+Connection reuse is configured through the `connectionSelector` configuration item. Users only need to create the
+connection resource once for reuse, improving connection management efficiency and simplifying the configuration
+process.
+
+1. **Create Resource**: As shown in the example below, create a connection with id `mqttcon1` via the API. The
+   connection parameters required can be configured in `props`. After the connection is successfully created, `mqttcon1`
+   can be found in the connection list API.
+
+    ```shell
+    POST http://localhost:9081/connections
+    {
+      "id": "mqttcon1"
+      "typ":"mqtt",
+      "props": {
+        server: "tcp://127.0.0.1:1883"
+      }
+    }
+    ```
+
+2. **Use in Data Source**: When configuring the MQTT source (`$ekuiper/etc/mqtt_source.yaml`), you can reference the
+   above connection configuration via `connectionSelector`, for example, both `demo_conf` and `demo2_conf` will
+   reference the connection configuration of `mqttcon1`.
+
+```yaml
+#Override the global configurations
+demo_conf: #Conf_key
+  qos: 0
+  connectionSelector: mqttcon1
+  servers: [ tcp://10.211.55.6:1883, tcp://127.0.0.1 ]
+
+#Override the global configurations
+demo2_conf: #Conf_key
+  qos: 0
+  connentionSelector: mqttcon1
+  servers: [ tcp://10.211.55.6:1883, tcp://127.0.0.1 ]
+```
+
+Based on `demo_conf` and `demo2_conf`, create two data streams `demo` and `demo2`:
+
+```text
+demo (
+    ...
+  ) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
+
+demo2 (
+    ...
+  ) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
+
+```
+
+When corresponding rules reference the above data streams, the source parts of the rules will share the connection.
+Here, `DATASOURCE` corresponds to the MQTT subscription topic, and the `qos` in the configuration item will be used as
+the `Qos` for subscription. In the example configuration above, `demo` subscribes to topic `test/` with Qos 0,
+and `demo2` subscribes to topic `test2/` with Qos 0.
+
+::: tip
+
+For MQTT sources, if two streams have the same `DATASOURCE` but different `qos` values, only the rule that starts first
+will trigger the subscription.
+
+:::
+
+You can also reuse the defined connection resource in the rule's action via `connentionSelector`.
+
+## Connection Status
+
+Connection status is divided into three types:
+
+1. **Connected**, represented by 1 in metrics.
+2. **Connecting**, represented by 0 in metrics.
+3. **Disconnected**, represented by 1 in metrics.
+
+Users can retrieve the connection status via the connection API. Additionally, users can view the connection status in
+the rule's source/sink metrics, for example, the `source_demo_0_connection_status` metric indicates the connection
+status of the `demo` stream. For a complete list of supported connection metrics, please refer to
+the [Metrics List](../../operation/usage/monitor_with_prometheus.md#metric-types).
diff --git a/docs/en_US/guide/connector.md b/docs/en_US/guide/connector.md
index af102e25..6642d823 100644
--- a/docs/en_US/guide/connector.md
+++ b/docs/en_US/guide/connector.md
@@ -73,79 +73,6 @@ For specialized data dispatch requirements or integrations with particular platf
 
 [Data templates](./sinks/data_template.md) in eKuiper allow for "secondary processing" of analysis results to cater to the diverse formatting requirements of different sink systems. Utilizing the Golang template system, eKuiper provides mechanisms for dynamic data transformation, conditional outputs, and iterative processing. This ensures compatibility and precise formatting for various sinks.
 
-## Connection Selector
-
-The connector selector is a powerful feature in eKuiper that allows users to define a connection once and reuse it across multiple configurations. It ensures efficient connection management and reduces redundancy.
-
-To define a global connection configuration, use the `connectionSelector` key to name your connection, e.g., `mqtt.localConnection`. Override global configurations with custom configurations but reference the same `connectionSelector`.
-
-For example, consider the configurations `demo_conf` and `demo2_conf`:
-
-```yaml
-#Override the global configurations
-demo_conf: #Conf_key
-  qos: 0
-  connectionSelector: mqtt.localConnection
-  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
-
-#Override the global configurations
-demo2_conf: #Conf_key
-  qos: 0
-  connentionSelector: mqtt.localConnection
-  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
-```
-
-Both configurations reference the same `connectionSelector`, indicating that they utilize the same MQTT connection. When streams `demo` and `demo2` are defined based on these configurations:
-
-```text
-demo (
-    ...
-  ) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
-
-demo2 (
-    ...
-  ) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
-
-```
-
-They inherently share the MQTT connection. Specifically:
-
-- The stream `demo` subscribes to the MQTT topic `test/` with a QoS of 0.
-- The stream `demo2` subscribes to `test2/`, also with a QoS of 0.
-
-::: tip
-
-For MQTT sources, if two streams have the same `DATASOURCE` but differing `qos` values, only the rule started first will trigger a subscription.
-
-:::
-
-**Configuration**
-
-The actual connection profiles, like `mqtt.localConnection`, are usually defined in a separate file, such as `connections/connection.yaml`.
-
-Example
-
-```yaml
-mqtt:
-  localConnection: #connection key
-    server: "tcp://127.0.0.1:1883"
-    username: ekuiper
-    password: password
-    #certificationPath: /var/kuiper/xyz-certificate.pem
-    #privateKeyPath: /var/kuiper/xyz-private.pem.ke
-    #insecureSkipVerify: false
-    #protocolVersion: 3
-    clientid: ekuiper
-  cloudConnection: #connection key
-    server: "tcp://broker.emqx.io:1883"
-    username: user1
-    password: password
-    #certificationPath: /var/kuiper/xyz-certificate.pem
-    #privateKeyPath: /var/kuiper/xyz-private.pem.ke
-    #insecureSkipVerify: false
-    #protocolVersion: 3
-```
-
 ## Batch Configuration
 
 For advanced data stream processing, eKuiper offers an array of connectors like Memory, File, MQTT, and more. To streamline the integration, eKuiper’s REST API introduces the capability for batch configuration, allowing users to simultaneously import or export multiple configurations.
diff --git a/docs/en_US/guide/rules/overview.md b/docs/en_US/guide/rules/overview.md
index 27e67422..38e9f984 100644
--- a/docs/en_US/guide/rules/overview.md
+++ b/docs/en_US/guide/rules/overview.md
@@ -301,3 +301,6 @@ When we try to send a record to the stream, the status of the rule is obtained a
 ```
 
 It can be seen that `records_in_total` and `records_out_total` of each operator have changed from 0 to 1, which means that the operator has received a record and passed a record to the next operator, and finally sent to the `sink` and the `sink` wrote 1 record.
+
+If Prometheus configuration is enabled, these metrics will also be collected by Prometheus. For a complete list of
+operational metrics, please refer to the [Metrics List](../../operation/usage/monitor_with_prometheus.md#metric-types).
diff --git a/docs/en_US/operation/usage/monitor_with_prometheus.md b/docs/en_US/operation/usage/monitor_with_prometheus.md
index df2bba2a..ee220d48 100644
--- a/docs/en_US/operation/usage/monitor_with_prometheus.md
+++ b/docs/en_US/operation/usage/monitor_with_prometheus.md
@@ -70,6 +70,14 @@ After version 1.6.1, we added two more exception-related metrics to facilitate t
 - last_exception: the error message of the last exception.
 - last_exception_time: the time of the last exception.
 
+After version 2.0.0, we added connection-related metrics for source/sink.
+
+- connection_status: Connection status. 1 for connected, 0 for connecting, -1 for disconnected.
+- connection_last_connected_time: The last successful connection time.
+- connection_last_disconnected_time: The last disconnection time.
+- connection_last_disconnected_message: The message of the last disconnection exception.
+- connection_last_try_time: The last reconnection attempt time.
+
 The numeric types of these metrics can all be monitored using Prometheus. In the next section we will describe how to configure the Prometheus service in eKuiper.
 
 ## Configuring the Prometheus Service in eKuiper
diff --git a/docs/en_US/operation/usage/trace_data.md b/docs/en_US/operation/usage/trace_data.md
index cbfdb8f5..ddaf9e84 100644
--- a/docs/en_US/operation/usage/trace_data.md
+++ b/docs/en_US/operation/usage/trace_data.md
@@ -8,6 +8,7 @@ You can expose Open Telemetry data to the remote Open Telemetry Collector throug
 
 ```yaml
 openTelemetry:
+  serviceName: kuiperd-service
   enableRemoteCollector: false
   remoteEndpoint: localhost:4318
   localTraceCapacity: 2048
diff --git a/docs/zh_CN/guide/connections/overview.md b/docs/zh_CN/guide/connections/overview.md
new file mode 100644
index 00000000..90794987
--- /dev/null
+++ b/docs/zh_CN/guide/connections/overview.md
@@ -0,0 +1,114 @@
+# 连接管理
+
+Source 和 Sink 用于与外部系统的交互，其中都会包含连接外部资源的动作。本章主要讲解 eKuiper 中对连接的管理。
+
+## 连接类型
+
+不同的连接类型复杂度各不相同，例如 MQTT 长连接需要关注连接的状态，规则运行中可能出现连接的断连，需要自动重连等复杂的管理；而
+HTTP 连接默认为无状态的连接，状态管理较为简单。为了统一管理复杂的连接资源创建，复用，自动重连以及获取连接状态等功能，eKuiper
+v2 增加了内部的连接池组件，并适配了一系列连接类型：
+
+- MQTT 连接
+- Neuron 连接
+- EdgeX 连接
+- SQL 连接
+- HTTP 连接 （包括 REST sink，HTTP Pull source，HTTP push source 使用的连接）
+- WebSocket 连接
+
+其余连接类型可能会在后续版本中陆续接入。接入连接池的连接类型可通过 API 进行资源的独立创建，并获取 API。
+
+eKuiper 中对各种连接的生命周期的管理分为 3 种：
+
+1. 连接附属于规则：默认情况下，连接由 Source/Sink 的实现自行管理，其生命周期由使用的规则进行控制。规则启动时，使用到的连接资源才会开始连接；规则结束时，连接将被关闭。在下例中，我们创建了
+   memory 类型的数据流 memStream。由于该类型未接入连接池，只有当使用该流的规则启动时，才会进行连接。
+
+    ```sql
+    create stream memStream () WITH (TYPE="mqtt", DATASOURCE="demo")
+    ```
+
+2. 连接池管理的匿名连接资源：部分连接类型适配了连接池管理接口，其生命周期由连接池管理。当启动包含这些类型连接的规则时，规则会从连接池获取匿名（实际资源
+   id 由规则生成，且不会被共享）资源。在下例中，我们创建了 mqtt 类型的数据流 mqttStream。连接为匿名连接，由于该类型适配了连接池，我们可以在连接
+   API 中获取到该连接。规则删除时，对应连接也会删除。
+
+    ```sql
+    create stream mqttStream () WITH (TYPE="mqtt", DATASOURCE="demo")
+    ```
+
+3. 用户创建的连接资源：用户可通过[连接管理API](../../api/restapi/connection.md) 进行资源的增删改查。通过 API 创建的资源必须指定唯一的
+   id，规则中可引用此处创建的规则资源。**请注意**：只有适配连接池的连接资源可通过 API
+   进行管理。这种连接类型创建的连接为独立的物理连接，创建完后会立即运行，无需依附于规则。它可以被多个规则，或者多个
+   source/sink 共用。
+
+**请注意**：用户创建的连接为实体连接，会自动重连直到连接成功为止。
+
+### 连接重用
+
+用户创建的连接资源可以独立运行，多个规则可以引用该命名资源。连接重用是通过 `connectionSelector`
+配置项进行配置。用户只需创建一次连接资源即可复用，提升连接管理效率，简化配置流程。
+
+1. 创建资源，如下例通过 API 创建 id 为 mqttcon1 的连接。可在 props 中配置连接需要的参数。连接创建成功后，mqttcon1 可在连接列表
+   API 中找到。
+
+    ```shell
+    POST http://localhost:9081/connections
+    {
+      "id": "mqttcon1"
+      "typ":"mqtt",
+      "props": {
+        server: "tcp://127.0.0.1:1883"
+      }
+    }
+    ```
+
+2. 在数据源中使用。在配置 MQTT 源（`$ekuiper/etc/mqtt_source.yaml`）时，可通过 `connectionSelector`
+   引用以上连接配置，例如`demo_conf` 和 `demo2_conf` 都将引用 `mqttcon1` 的连接配置。
+
+```yaml
+#Override the global configurations
+demo_conf: #Conf_key
+   qos: 0
+   connectionSelector: mqttcon1
+   servers: [ tcp://10.211.55.6:1883, tcp://127.0.0.1 ]
+
+#Override the global configurations
+demo2_conf: #Conf_key
+   qos: 0
+   connentionSelector: mqttcon1
+   servers: [ tcp://10.211.55.6:1883, tcp://127.0.0.1 ]
+```
+
+基于 `demo_conf` 和 `demo2_conf` 分别创建两个数据流 `demo` 和 `demo2`：
+
+```text
+demo (
+    ...
+  ) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
+
+demo2 (
+    ...
+  ) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
+
+```
+
+当相应的规则分别引用以上数据流时，规则之间的源部分将共享连接。在这里 `DATASOURCE` 对应 mqtt 订阅的 topic，配置项中的 `qos`
+将用作订阅时的 `Qos`。在以上示例配置中，`demo` 以 Qos 0 订阅 topic `test/`，`demo2` 以 Qos 0 订阅 topic `test2/` 。
+
+::: tip
+
+对于MQTT源，如果两个流具有相同的 `DATASOURCE` 但 `qos` 值不同，则只有先启动的规则才会触发订阅。
+
+:::
+
+也可以在规则的 action 中，通过 connentionSelector 重用定义的连接资源。
+
+## 连接状态
+
+连接状态分成 3 种：
+
+1. 已连接，指标中用 1 表示。
+2. 连接中，指标中用 0 表示。
+3. 未连接，指标中用 1 表示。
+
+用户可通过连接 API 获取连接的状态。同时，用户也可通过规则的指标查看规则 source/sink
+中连接的状态，例如 `source_demo_0_connection_status` 指标表示 demo
+流的连接状态。所有支持的连接指标请查看[指标列表](../../operation/usage/monitor_with_prometheus.md#运行指标)。
diff --git a/docs/zh_CN/guide/connector.md b/docs/zh_CN/guide/connector.md
index 20eec63d..9e055a39 100644
--- a/docs/zh_CN/guide/connector.md
+++ b/docs/zh_CN/guide/connector.md
@@ -73,72 +73,6 @@ eKuiper Sink 连接器负责将 eKuiper 处理后的数据发送到各种目标
 
 eKuiper [数据模板](./sinks/data_template.md) 支持用户对分析结果进行"二次处理"，以满足不同接收系统的多样化格式要求。利用 Golang 模板系统，eKuiper 提供了动态数据转换、条件输出和迭代处理的机制，确保了与各种接收器的兼容性和精确格式化。
 
-## 连接器的重用
-
-eKuiper 支持通过 `connectionSelector` 配置项对连接器进行重用，用户只需一次定义即可在多个配置中重用，提升连接管理效率，简化配置流程。
-
-**配置**
-
-以 MQTT 数据源为例，您可首先在连接配置文件 `connections/connection.yaml` 中定义 MQTT 全局连接信息，例如  `mqtt.localConnection` 和 `mqtt.cloudConnection`。
-
-```yaml
-mqtt:
-  localConnection: #connection key
-    server: "tcp://127.0.0.1:1883"
-    username: ekuiper
-    password: password
-    #certificationPath: /var/kuiper/xyz-certificate.pem
-    #privateKeyPath: /var/kuiper/xyz-private.pem.ke
-    #insecureSkipVerify: false
-    #protocolVersion: 3
-    clientid: ekuiper
-  cloudConnection: #connection key
-    server: "tcp://broker.emqx.io:1883"
-    username: user1
-    password: password
-    #certificationPath: /var/kuiper/xyz-certificate.pem
-    #privateKeyPath: /var/kuiper/xyz-private.pem.ke
-    #insecureSkipVerify: false
-    #protocolVersion: 3
-```
-
-在配置 MQTT 源（`$ekuiper/etc/mqtt_source.yaml`）时，可通过 `connectionSelector` 引用以上连接配置，例如`demo_conf` 和 `demo2_conf` 都将引用 `mqtt.localConnection` 的连接配置。
-
-```yaml
-#Override the global configurations
-demo_conf: #Conf_key
-  qos: 0
-  connectionSelector: mqtt.localConnection
-  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
-
-#Override the global configurations
-demo2_conf: #Conf_key
-  qos: 0
-  connentionSelector: mqtt.localConnection
-  servers: [tcp://10.211.55.6:1883, tcp://127.0.0.1]
-```
-
-基于 `demo_conf` 和 `demo2_conf` 分别创建两个数据流 `demo` 和 `demo2`：
-
-```text
-demo (
-    ...
-  ) WITH (DATASOURCE="test/", FORMAT="JSON", CONF_KEY="demo_conf");
-
-demo2 (
-    ...
-  ) WITH (DATASOURCE="test2/", FORMAT="JSON", CONF_KEY="demo2_conf");
-
-```
-
-当相应的规则分别引用以上数据流时，规则之间的源部分将共享连接。在这里 `DATASOURCE` 对应 mqtt 订阅的 topic，配置项中的 `qos` 将用作订阅时的 `Qos`。在以上示例配置中，`demo` 以 Qos 0 订阅 topic `test/`，`demo2` 以 Qos 0 订阅 topic `test2/` 。
-
-::: tip
-
-对于MQTT源，如果两个流具有相同的 `DATASOURCE` 但 `qos` 值不同，则只有先启动的规则才会触发订阅。
-
-:::
-
 ## 批量配置
 
 eKuiper 提供了 Memory、File、MQTT 等多种数据连接器。为进一步简化用户的配置流程，eKuiper 通过 REST API 引入了批量配置功能，支持用户同时导入或导出多个配置。
diff --git a/docs/zh_CN/guide/rules/overview.md b/docs/zh_CN/guide/rules/overview.md
index c04be75c..69c466bc 100644
--- a/docs/zh_CN/guide/rules/overview.md
+++ b/docs/zh_CN/guide/rules/overview.md
@@ -283,3 +283,6 @@ cronDatetimeRange 支持结构体数组，你可以声明一组时间段来表
 ```
 
 可以看到每个算子的 `records_in_total` 与 `records_out_total` 都由 0 变为了 1，代表了该算子接收到了一条记录，并向下一个算子传递了一条记录，最终在 `sink` 端向 sink 写入了 1 条记录。
+
+若开启 Prometheus 配置，这些指标也会收集到 Prometheus
+中。全部的运行指标列表请查看[指标列表](../../operation/usage/monitor_with_prometheus.md#运行指标)。
diff --git a/docs/zh_CN/operation/usage/monitor_with_prometheus.md b/docs/zh_CN/operation/usage/monitor_with_prometheus.md
index 4a625637..4d52cbf7 100644
--- a/docs/zh_CN/operation/usage/monitor_with_prometheus.md
+++ b/docs/zh_CN/operation/usage/monitor_with_prometheus.md
@@ -70,6 +70,14 @@ kuiper_rule_count: eKuiper 中有多少条规则运行，多少条规则暂停
 - last_exception：最近一次的异常的错误信息。
 - last_exception_time：最近一次异常的发生时间。
 
+在 2.0.0 版本之后，我们为 source/sink 添加了连接相关指标。
+
+- connection_status：连接状态。1 为已连接，0 为连接中，-1 为未连接。
+- connection_last_connected_time：最近一次连接成功时间
+- connection_last_disconnected_time：最近一次断连时间
+- connection_last_disconnected_message：最近一次断连异常的消息
+- connection_last_try_time：最近一次重连时间
+
 这些运行指标中的数值类型指标均可使用 Prometheus 进行监控。下一节我们将描述如何配置 eKuiper 中的 Prometheus 服务。
 
 ## 配置 eKuiper 的 Prometheus 服务
diff --git a/docs/zh_CN/operation/usage/trace_data.md b/docs/zh_CN/operation/usage/trace_data.md
index b035d3a5..8c29ea11 100644
--- a/docs/zh_CN/operation/usage/trace_data.md
+++ b/docs/zh_CN/operation/usage/trace_data.md
@@ -8,6 +8,7 @@ eKuiper 的规则是一个持续运行的流式计算任务。规则用于处理
 
 ```yaml
 openTelemetry:
+  serviceName: kuiperd-service
   enableRemoteCollector: false
   remoteEndpoint: localhost:4318
   localTraceCapacity: 2048
diff --git a/etc/kuiper.yaml b/etc/kuiper.yaml
index 36581793..15749431 100644
--- a/etc/kuiper.yaml
+++ b/etc/kuiper.yaml
@@ -142,6 +142,7 @@ portable:
   initTimeout: 5s
 
 openTelemetry:
+  serviceName: kuiperd-service
   enableRemoteCollector: false
   remoteEndpoint: localhost:4318
   localTraceCapacity: 2048
diff --git a/internal/io/file/source.go b/internal/io/file/source.go
index 26db5c69..8161d61f 100644
--- a/internal/io/file/source.go
+++ b/internal/io/file/source.go
@@ -60,7 +60,9 @@ type Source struct {
 	isDir  bool
 	config *SourceConfig
 	reader modules.FileStreamReader
-	eof    api.EOFIngest
+	// attach to a reader
+	decorator modules.FileStreamDecorator
+	eof       api.EOFIngest
 }
 
 func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
@@ -140,6 +142,14 @@ func (fs *Source) Provision(ctx api.StreamContext, props map[string]any) error {
 		}
 	}
 	fs.config = cfg
+	decorator, ok := modules.GetFileStreamDecorator(ctx, cfg.FileType)
+	if ok {
+		err = decorator.Provision(ctx, props)
+		if err != nil {
+			return err
+		}
+		fs.decorator = decorator
+	}
 	return nil
 }
 
@@ -230,7 +240,7 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 		maxSize = int(info.Size())
 	}
 	if fs.config.IgnoreStartLines > 0 || fs.config.IgnoreEndLines > 0 {
-		r = ignoreLines(ctx, r, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
+		r = ignoreLines(ctx, r, fs.decorator, fs.config.IgnoreStartLines, fs.config.IgnoreEndLines)
 	}
 	if closer, ok := r.(io.Closer); ok {
 		defer func() {
@@ -255,6 +265,9 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 				break
 			}
 			rcvTime := timex.GetNow()
+			if fs.decorator != nil {
+				line = fs.decorator.Decorate(ctx, line)
+			}
 			ingest(ctx, line, meta, rcvTime)
 			if fs.config.SendInterval > 0 {
 				time.Sleep(fs.config.SendInterval)
@@ -289,7 +302,7 @@ func (fs *Source) parseFile(ctx api.StreamContext, file string, ingest api.Tuple
 	}
 }
 
-func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int, ignoreEndLines int) io.Reader {
+func ignoreLines(ctx api.StreamContext, reader io.Reader, decorator modules.FileStreamDecorator, ignoreStartLines int, ignoreEndLines int) io.Reader {
 	r, w := io.Pipe()
 	go func() {
 		e := infra.SafeRun(func() error {
@@ -305,6 +318,10 @@ func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int,
 			tempLines := make([][]byte, 0, ignoreEndLines)
 			for scanner.Scan() {
 				if ln >= ignoreStartLines {
+					if ln == ignoreStartLines && decorator != nil {
+						// Send EOF to decorator
+						decorator.ReadMeta(ctx, nil)
+					}
 					if ignoreEndLines > 0 { // the last n line are left in the tempLines
 						slot := (ln - ignoreStartLines) % ignoreEndLines
 						if len(tempLines) <= slot { // first round
@@ -334,6 +351,10 @@ func ignoreLines(ctx api.StreamContext, reader io.Reader, ignoreStartLines int,
 							break
 						}
 					}
+				} else {
+					if decorator != nil {
+						decorator.ReadMeta(ctx, scanner.Bytes())
+					}
 				}
 				ln++
 			}
diff --git a/pkg/cast/time.go b/pkg/cast/time.go
index e4a1fae6..e829e09a 100644
--- a/pkg/cast/time.go
+++ b/pkg/cast/time.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -138,7 +138,18 @@ func ParseTime(t string, f string) (_ time.Time, err error) {
 		TimeFormats:  now.TimeFormats,
 	}
 	if f != "" {
-		c.TimeFormats = []string{f}
+		c.TimeFormats = append([]string{f}, c.TimeFormats...)
+	}
+	return c.Parse(t)
+}
+
+func ParseTimeByFormats(t string, formats []string) (_ time.Time, err error) {
+	c := &now.Config{
+		TimeLocation: localTimeZone,
+		TimeFormats:  now.TimeFormats,
+	}
+	if len(formats) > 0 {
+		c.TimeFormats = append(formats, c.TimeFormats...)
 	}
 	return c.Parse(t)
 }
diff --git a/pkg/modules/file.go b/pkg/modules/file.go
index 880fdc48..6a9a74f0 100644
--- a/pkg/modules/file.go
+++ b/pkg/modules/file.go
@@ -63,9 +63,38 @@ func RegisterFileStreamReader(name string, provider FileStreamReaderProvider) {
 	fileStreamReaders[name] = provider
 }
 
+func RegisterFileStreamReaderAlias(alias string, ref string) {
+	fileStreamReaders[alias] = fileStreamReaders[ref]
+}
+
 func GetFileStreamReader(ctx api.StreamContext, name string) (FileStreamReader, bool) {
 	if p, ok := fileStreamReaders[name]; ok {
 		return p(ctx), true
 	}
 	return nil, false
 }
+
+type FileStreamDecorator interface {
+	// Provision Set up the static properties
+	Provision(ctx api.StreamContext, props map[string]any) error
+	// ReadMeta Read the metadata from the file source, and save in the decorator itself
+	// It will receive lines, when receiving EOF, there will be no more lines.
+	ReadMeta(ctx api.StreamContext, line []byte)
+	// Decorate the file source
+	Decorate(ctx api.StreamContext, data any) any
+}
+
+type FileStreamDecoratorProvider func(ctx api.StreamContext) FileStreamDecorator
+
+var fileStreamDecorators = map[string]FileStreamDecoratorProvider{}
+
+func RegisterFileStreamDecorator(name string, provider FileStreamDecoratorProvider) {
+	fileStreamDecorators[name] = provider
+}
+
+func GetFileStreamDecorator(ctx api.StreamContext, name string) (FileStreamDecorator, bool) {
+	if p, ok := fileStreamDecorators[name]; ok {
+		return p(ctx), true
+	}
+	return nil, false
+}
diff --git a/pkg/tracer/tracer.go b/pkg/tracer/tracer.go
index fa95fc10..6fc2d7a9 100644
--- a/pkg/tracer/tracer.go
+++ b/pkg/tracer/tracer.go
@@ -77,6 +77,7 @@ func (g *GlobalTracerManager) SetTracer(enableRemote bool, serviceName, endpoint
 	if err != nil {
 		return err
 	}
+	g.SpanExporter = exporter
 	opts = append(opts, sdktrace.WithBatcher(exporter))
 	tp := sdktrace.NewTracerProvider(opts...)
 	otel.SetTracerProvider(tp)


Test patch:
diff --git a/internal/io/file/source_decorator_test.go b/internal/io/file/source_decorator_test.go
new file mode 100644
index 00000000..e43a9ede
--- /dev/null
+++ b/internal/io/file/source_decorator_test.go
@@ -0,0 +1,87 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed 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 file
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type MockDecoder struct {
+	meta string
+}
+
+func (m *MockDecoder) Provision(_ api.StreamContext, props map[string]any) error {
+	_, ok := props["wrong"]
+	if ok {
+		return fmt.Errorf("wrong")
+	}
+	return nil
+}
+
+func (m *MockDecoder) ReadMeta(_ api.StreamContext, line []byte) {
+	if line != nil {
+		m.meta = string(line)
+		fmt.Println(m.meta)
+	}
+}
+
+func (m *MockDecoder) Decorate(ctx api.StreamContext, data any) any {
+	data.(map[string]any)["meta"] = m.meta
+	return data
+}
+
+func TestCSVDec(t *testing.T) {
+	path, err := os.Getwd()
+	if err != nil {
+		t.Fatal(err)
+	}
+	path = filepath.Join(path, "test", "csv")
+
+	modules.RegisterFileStreamReaderAlias("mockmock", "csv")
+	modules.RegisterFileStreamDecorator("mockmock", func(ctx api.StreamContext) modules.FileStreamDecorator {
+		return &MockDecoder{}
+	})
+
+	meta := map[string]any{
+		"file": filepath.Join(path, "a.csv"),
+	}
+	mc := timex.Clock
+	exp := []api.MessageTuple{
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "1", "ts": "1670170500", "value": "161.927872", "meta": "<special content>"}, meta, mc.Now()),
+		model.NewDefaultSourceTuple(map[string]any{"@": "#", "id": "2", "ts": "1670170900", "value": "176", "meta": "<special content>"}, meta, mc.Now()),
+	}
+	r := GetSource()
+	mock.TestSourceConnector(t, r, map[string]any{
+		"path":             path,
+		"fileType":         "mockmock",
+		"datasource":       "a.csv",
+		"hasHeader":        true,
+		"delimiter":        "\t",
+		"ignoreStartLines": 3,
+		"ignoreEndLines":   1,
+	}, exp, func() {
+		// do nothing
+	})
+}
diff --git a/pkg/cast/time_test.go b/pkg/cast/time_test.go
index 40a8a6f1..81174ec6 100644
--- a/pkg/cast/time_test.go
+++ b/pkg/cast/time_test.go
@@ -301,3 +301,15 @@ func TestConvertFormat(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(1718024079657497900), d.UnixNano())
 }
+
+func TestParseTimeFormats(t *testing.T) {
+	err := SetTimeZone("UTC")
+	require.NoError(t, err)
+	tsstr := "2023-12-16 00:15"
+	tts, err := ParseTimeByFormats(tsstr, []string{"2006-01-02 15:04", "2006-01-02 15-04-05"})
+	require.NoError(t, err)
+
+	timeString := "2023-12-16T00:15:00Z"
+	tt, _ := time.Parse(time.RFC3339, timeString)
+	require.Equal(t, tt, tts)
+}
diff --git a/pkg/tracer/tracer_test.go b/pkg/tracer/tracer_test.go
index 488322d0..43f88a4e 100644
--- a/pkg/tracer/tracer_test.go
+++ b/pkg/tracer/tracer_test.go
@@ -34,3 +34,11 @@ func TestSaveLoadTracerConfig(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, newConfig, traceConfig)
 }
+
+func TestSetTracer(t *testing.T) {
+	conf.IsTesting = true
+	conf.InitConf()
+	err := SetTracer(&TracerConfig{})
+	require.NoError(t, err)
+	require.NotNil(t, globalTracerManager.SpanExporter)
+}