feat: add config ingester_traffic_overflow_action

Author: lzf575

agent/crates/public/src/queue/overwrite_queue.rs

@@ -55,6 +55,8 @@ struct OverwriteQueue<T: Sized> {
 
     counter: Counter,
 
+    total_overwritten_count: AtomicU64,
+
     _marker: PhantomData<T>,
 }
 
@@ -78,6 +80,7 @@ impl<T> OverwriteQueue<T> {
             notify: Condvar::new(),
             terminated: AtomicBool::new(false),
             counter: Counter::default(),
+            total_overwritten_count: AtomicU64::new(0),
             _marker: PhantomData,
         }
     }
@@ -130,6 +133,8 @@ impl<T> OverwriteQueue<T> {
                 self.counter
                     .overwritten
                     .fetch_add(to_overwrite as u64, Ordering::Relaxed);
+                self.total_overwritten_count
+                    .fetch_add(to_overwrite as u64, Ordering::Relaxed);
             }
         }
         let free_after_end = self.size - (raw_end & (self.size - 1));
@@ -416,6 +421,13 @@ impl<T> Receiver<T> {
             }
         }
     }
+
+    pub fn total_overwritten_count(&self) -> u64 {
+        self.counter()
+            .queue
+            .total_overwritten_count
+            .load(Ordering::Relaxed)
+    }
 }
 
 impl<T> Drop for Receiver<T> {

agent/src/config/config.rs

@@ -2129,6 +2129,7 @@ pub struct Communication {
     #[serde(deserialize_with = "deser_usize_with_mega_unit")]
     pub grpc_buffer_size: usize,
     pub max_throughput_to_ingester: u64,
+    pub ingester_traffic_overflow_action: u8, // 0: wait, 1: drop
     pub request_via_nat_ip: bool,
     pub proxy_controller_ip: String,
     pub proxy_controller_port: u16,
@@ -2145,6 +2146,7 @@ impl Default for Communication {
             ingester_port: 30033,
             grpc_buffer_size: 5 << 20,
             max_throughput_to_ingester: 100,
+            ingester_traffic_overflow_action: 0,
             request_via_nat_ip: false,
         }
     }

agent/src/config/handler.rs

@@ -231,7 +231,8 @@ pub struct SenderConfig {
     pub npb_bps_threshold: u64,
     pub npb_socket_type: agent::SocketType,
     pub multiple_sockets_to_ingester: bool,
-    pub max_throughput_to_ingester: u64, // unit: Mbps
+    pub max_throughput_to_ingester: u64,      // unit: Mbps
+    pub ingester_traffic_overflow_action: u8, // 0: wait, 1: drop
     pub collector_socket_type: agent::SocketType,
     pub standalone_data_file_size: u32,
     pub standalone_data_file_dir: String,
@@ -1740,6 +1741,10 @@ impl TryFrom<(Config, UserConfig)> for ModuleConfig {
                     .throughput_monitoring_interval,
                 multiple_sockets_to_ingester: conf.outputs.socket.multiple_sockets_to_ingester,
                 max_throughput_to_ingester: conf.global.communication.max_throughput_to_ingester,
+                ingester_traffic_overflow_action: conf
+                    .global
+                    .communication
+                    .ingester_traffic_overflow_action,
                 collector_socket_type: conf.outputs.socket.data_socket_type,
                 standalone_data_file_size: conf.global.standalone_mode.max_data_file_size,
                 standalone_data_file_dir: conf.global.standalone_mode.data_file_dir.clone(),
@@ -3837,6 +3842,17 @@ impl ConfigHandler {
             );
             communication.max_throughput_to_ingester = new_communication.max_throughput_to_ingester;
         }
+        if communication.ingester_traffic_overflow_action
+            != new_communication.ingester_traffic_overflow_action
+        {
+            info!(
+                "Update global.communication.ingester_traffic_overflow_action from {:?} to {:?}.",
+                communication.ingester_traffic_overflow_action,
+                new_communication.ingester_traffic_overflow_action
+            );
+            communication.ingester_traffic_overflow_action =
+                new_communication.ingester_traffic_overflow_action;
+        }
         if communication.ingester_ip != new_communication.ingester_ip {
             info!(
                 "Update global.communication.ingester_ip from {:?} to {:?}.",

agent/src/sender/uniform_sender.rs

@@ -53,6 +53,7 @@ pub struct SenderCounter {
     pub tx: AtomicU64,
     pub tx_bytes: AtomicU64,
     pub dropped: AtomicU64,
+    pub waited: AtomicU64,
 }
 
 impl RefCountable for SenderCounter {
@@ -89,6 +90,11 @@ impl RefCountable for SenderCounter {
                 CounterType::Counted,
                 CounterValue::Unsigned(self.dropped.swap(0, Ordering::Relaxed)),
             ),
+            (
+                "waited",
+                CounterType::Counted,
+                CounterValue::Unsigned(self.waited.swap(0, Ordering::Relaxed)),
+            ),
         ]
     }
 }
@@ -380,6 +386,7 @@ pub struct UniformSender<T> {
 
     input: Arc<Receiver<T>>,
     counter: Arc<SenderCounter>,
+    overwritten_count: u64,
 
     encoder: Encoder<T>,
     private_conn: Mutex<Connection>,
@@ -426,6 +433,7 @@ impl<T: Sendable> UniformSender<T> {
             name,
             input,
             counter: Arc::new(SenderCounter::default()),
+            overwritten_count: 0,
             encoder: Encoder::new(
                 0,
                 SendMessageType::TaggedFlow,
@@ -626,7 +634,11 @@ impl<T: Sendable> UniformSender<T> {
         }
     }
 
-    fn is_exceed_max_throughput(&mut self, max_throughput_mbps: u64) -> bool {
+    fn is_exceed_max_throughput(
+        &mut self,
+        max_throughput_mbps: u64,
+        ingester_traffic_overflow_action: u8,
+    ) -> bool {
         if max_throughput_mbps == 0 {
             return false;
         }
@@ -637,7 +649,7 @@ impl<T: Sendable> UniformSender<T> {
                 .unwrap();
 
             let used = now - Duration::from_nanos(SENT_START_DURATION.load(Ordering::Relaxed));
-            if used > Duration::from_secs(1) {
+            if used >= Duration::from_secs(1) {
                 SENT_START_DURATION.store(now.as_nanos() as u64, Ordering::Relaxed);
                 TOTAL_SENT_BYTES.store(0, Ordering::Relaxed);
             } else {
@@ -646,11 +658,23 @@ impl<T: Sendable> UniformSender<T> {
                     > Duration::from_secs(5)
                 {
                     warn!(
-                        "{} sender dropping message, throughput execeed setting value 'max_throughput_to_ingester' {}Mbps",
-                        self.name, max_throughput_mbps
+                        "{} sender dropping message, throughput execeed setting value 'max_throughput_to_ingester' {}Mbps, action {} (0: wait, 1: drop), total overwrittern count {}",
+                        self.name, max_throughput_mbps, ingester_traffic_overflow_action, self.overwritten_count
                     );
                     LAST_LOGGING_DURATION.store(now.as_nanos() as u64, Ordering::Relaxed);
                 }
+                // action is wait
+                if ingester_traffic_overflow_action == 0 {
+                    thread::sleep(Duration::from_secs(1) - used);
+                    // if the queue loses data, need to report the exception
+                    if self.input.total_overwritten_count() > self.overwritten_count {
+                        self.exception_handler
+                            .set(Exception::DataBpsThresholdExceeded);
+                        self.overwritten_count = self.input.total_overwritten_count();
+                    }
+                    return true;
+                }
+                // action is drop
                 self.exception_handler
                     .set(Exception::DataBpsThresholdExceeded);
                 return true;
@@ -678,6 +702,7 @@ impl<T: Sendable> UniformSender<T> {
             let config = self.config.load();
             let socket_type = config.collector_socket_type;
             let max_throughput_mpbs = config.max_throughput_to_ingester;
+            let ingester_traffic_overflow_action = config.ingester_traffic_overflow_action;
             match self.input.recv_all(
                 &mut batch,
                 Some(Duration::from_secs(Self::QUEUE_READ_TIMEOUT)),
@@ -688,12 +713,21 @@ impl<T: Sendable> UniformSender<T> {
                         start_cached = Instant::now();
                         self.cached = false;
                     }
-                    if self.is_exceed_max_throughput(max_throughput_mpbs) {
-                        self.counter
-                            .dropped
-                            .fetch_add(batch.len() as u64, Ordering::Relaxed);
-                        batch.clear();
-                        continue;
+                    if self.is_exceed_max_throughput(
+                        max_throughput_mpbs,
+                        ingester_traffic_overflow_action,
+                    ) {
+                        if ingester_traffic_overflow_action == 0 {
+                            self.counter
+                                .waited
+                                .fetch_add(batch.len() as u64, Ordering::Relaxed);
+                        } else {
+                            self.counter
+                                .dropped
+                                .fetch_add(batch.len() as u64, Ordering::Relaxed);
+                            batch.clear();
+                            continue;
+                        }
                     }
                     for send_item in batch.drain(..) {
                         if !self.running.load(Ordering::Relaxed) {

server/agent_config/README-CH.md

@@ -1051,9 +1051,43 @@ global:
 **详细描述**:
 
 向 Server 端 Ingester 模块发送可观测性数据的最大允许流量,
-超过此限速时数据将会主动丢弃、且采集器会标记为异常状态并触发告警。
+若`ingester_traffic_overflow_action` 配置为`丢弃`，超过此限速时数据将会主动丢弃、且采集器会标记为异常状态并触发告警。
 配置为 0 表示不限速。
 
+### Ingester 流量超限的动作 {#global.communication.ingester_traffic_overflow_action}
+
+**标签**:
+
+`hot_update`
+
+**FQCN**:
+
+`global.communication.ingester_traffic_overflow_action`
+
+**默认值**:
+```yaml
+global:
+  communication:
+    ingester_traffic_overflow_action: 0
+```
+
+**枚举可选值**:
+| Value | Note                         |
+| ----- | ---------------------------- |
+| 0 | 等待 |
+| 1 | 丢弃 |
+
+**模式**:
+| Key  | Value                        |
+| ---- | ---------------------------- |
+| Type | int |
+
+**详细描述**:
+
+Ingester 流量超限的动作
+- 等待：暂停发送，数据缓存到队列，等待下次发送。
+- 丢弃：直接丢弃数据，并触发 Agent `数据流量达到限速`异常。
+
 ### 请求 NAT IP 地址 {#global.communication.request_via_nat_ip}
 
 **标签**:

server/agent_config/README.md

@@ -1075,10 +1075,45 @@ global:
 **Description**:
 
 The maximum allowed flow rate for sending observability data to the server-side Ingester module.
-When this rate limit is exceeded, the data will be actively discarded,
+If `ingester_traffic_overflow_action` is set to `drop`,
+when this rate limit is exceeded, the data will be actively discarded,
 and the agent will be marked as abnormal and trigger an alarm.
 Setting it to 0 means no speed limit.
 
+### Action when the Ingester traffic exceeds the limit {#global.communication.ingester_traffic_overflow_action}
+
+**Tags**:
+
+`hot_update`
+
+**FQCN**:
+
+`global.communication.ingester_traffic_overflow_action`
+
+**Default value**:
+```yaml
+global:
+  communication:
+    ingester_traffic_overflow_action: 0
+```
+
+**Enum options**:
+| Value | Note                         |
+| ----- | ---------------------------- |
+| 0 | wait |
+| 1 | drop |
+
+**Schema**:
+| Key  | Value                        |
+| ---- | ---------------------------- |
+| Type | int |
+
+**Description**:
+
+Action when the Ingester traffic exceeds the limit
+- wait: pause sending, cache data into queue, and wait for next sending
+- drop: the data is discarded directly and the Agent `DATA_BPS_THRESHOLD_EXCEEDED` exception is triggered
+
 ### Request via NAT IP Address {#global.communication.request_via_nat_ip}
 
 **Tags**:

server/agent_config/template.yaml

@@ -713,14 +713,40 @@ global:
     # description:
     #   en: |-
     #     The maximum allowed flow rate for sending observability data to the server-side Ingester module.
-    #     When this rate limit is exceeded, the data will be actively discarded,
+    #     If `ingester_traffic_overflow_action` is set to `drop`,
+    #     when this rate limit is exceeded, the data will be actively discarded,
     #     and the agent will be marked as abnormal and trigger an alarm.
     #     Setting it to 0 means no speed limit.
     #   ch: |-
     #     向 Server 端 Ingester 模块发送可观测性数据的最大允许流量,
-    #     超过此限速时数据将会主动丢弃、且采集器会标记为异常状态并触发告警。
+    #     若`ingester_traffic_overflow_action` 配置为`丢弃`，超过此限速时数据将会主动丢弃、且采集器会标记为异常状态并触发告警。
     #     配置为 0 表示不限速。
     max_throughput_to_ingester: 100
+    # type: int
+    # name:
+    #   en: Action when the Ingester traffic exceeds the limit
+    #   ch: Ingester 流量超限的动作
+    # unit:
+    # range: []
+    # enum_options:
+    #   - 0:
+    #       en: wait
+    #       ch: 等待
+    #   - 1:
+    #       en: drop
+    #       ch: 丢弃
+    # modification: hot_update
+    # ee_feature: false
+    # description:
+    #   en: |-
+    #     Action when the Ingester traffic exceeds the limit
+    #     - wait: pause sending, cache data into queue, and wait for next sending
+    #     - drop: the data is discarded directly and the Agent `DATA_BPS_THRESHOLD_EXCEEDED` exception is triggered
+    #   ch: |-
+    #     Ingester 流量超限的动作
+    #     - 等待：暂停发送，数据缓存到队列，等待下次发送。
+    #     - 丢弃：直接丢弃数据，并触发 Agent `数据流量达到限速`异常。
+    ingester_traffic_overflow_action: 0
     # type: bool
     # name:
     #   en: Request via NAT IP Address