mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 08:09:22 +00:00
fix: update remote wal config (#8359)
### **PR Type**
Enhancement, Bug fix
___
### **Description**
- Add two new pipeline config options
- Set safe defaults when zero provided
- Improve WAL pipeline timing controls
___
### Diagram Walkthrough
```mermaid
flowchart LR
CFG["Config struct (Pipeline)"] -- "add fields" --> F1["pipeline_file_push_back_interval (secs)"]
CFG -- "add fields" --> F2["pipeline_sink_task_spawn_interval_ms (ms)"]
VALID["check_pipeline_config"] -- "apply defaults if 0" --> F1
VALID -- "apply defaults if 0" --> F2
```
<details> <summary><h3> File Walkthrough</h3></summary>
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
<td>
<details>
<summary><strong>config.rs</strong><dd><code>New pipeline timing fields
and defaulting logic</code>
</dd></summary>
<hr>
src/config/src/config.rs
<ul><li>Add <code>pipeline_file_push_back_interval</code> with env
config.<br> <li> Add <code>pipeline_sink_task_spawn_interval_ms</code>
with env config.<br> <li> Enforce non-zero defaults in
<code>check_pipeline_config</code>.<br> <li> Preserve existing WAL size
handling.</ul>
</details>
</td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8359/files#diff-9aeee4a74c0520daa40f9b81c7016f5b6b3254375b968dd7598e745630980d62">+20/-0</a>
</td>
</tr>
</table></td></tr></tr></tbody></table>
</details>
___
This commit is contained in:
@@ -1936,6 +1936,18 @@ pub struct Pipeline {
|
||||
help = "pipeline max file retention time in seconds"
|
||||
)]
|
||||
pub pipeline_max_file_retention_time_seconds: u64,
|
||||
#[env_config(
|
||||
name = "ZO_PIPELINE_FILE_PUSH_BACK_INTERVAL",
|
||||
default = 2,
|
||||
help = "duration in seconds to push the file to back to the queue after a read complete"
|
||||
)]
|
||||
pub pipeline_file_push_back_interval: u64,
|
||||
#[env_config(
|
||||
name = "ZO_PIPELINE_SINK_TASK_SPAWN_INTERVAL_MS",
|
||||
default = 100,
|
||||
help = "interval in milliseconds to spawn a new sink task"
|
||||
)]
|
||||
pub pipeline_sink_task_spawn_interval_ms: u64,
|
||||
}
|
||||
|
||||
#[derive(EnvConfig, Default)]
|
||||
@@ -2781,6 +2793,14 @@ fn check_pipeline_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
|
||||
} else {
|
||||
cfg.pipeline.wal_size_limit *= 1024 * 1024;
|
||||
}
|
||||
|
||||
if cfg.pipeline.pipeline_file_push_back_interval == 0 {
|
||||
cfg.pipeline.pipeline_file_push_back_interval = 2; // 2 seconds
|
||||
}
|
||||
|
||||
if cfg.pipeline.pipeline_sink_task_spawn_interval_ms == 0 {
|
||||
cfg.pipeline.pipeline_sink_task_spawn_interval_ms = 100; // 100 milliseconds
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user