@cryptotaxi247 / netdata-1 / commits / 09cbde151

Refactor function progress: move timeout and progress authority to the plugin runtime and agent (#21723)

* Refactor function progress: move timeout authority to the agent The plugin runtime now owns progress reporting and the agent is the authoritative source for transaction timeouts. Function handlers only need to update atomic progress counters and react to cancellation — they no longer manage their own timeouts or emit progress responses directly. Key changes: Plugin runtime (netdata-plugin/rt): - Add ProgressState with atomic done/total counters that handlers can update from any context (async, spawn_blocking, rayon). - Add FunctionCallContext carrying ProgressState and a CancellationToken, passed to on_call() instead of a bare transaction ID. - Spawn a per-transaction ticker task that reads the atomic counters once per second and emits FUNCTION_PROGRESS responses to the agent. - Spawn a dedicated writer task for stdout so outbound I/O never blocks the main stdin select loop. - Remove on_cancellation() and on_progress() from the FunctionHandler trait; the runtime handles cancellation generically. Protocol (netdata-plugin/protocol): - Split FunctionProgress into FunctionProgressRequest (agent→plugin) and FunctionProgressResponse (plugin→agent, carries done/all counters). - Make FUNCTION_PROGRESS parsing direction-aware: input context parses the request form, output context parses the response form. - Implement encoding for FunctionProgressResponse. Journal-engine: - Replace Timeout with CancellationToken in batch_compute_file_indexes and add an optional progress_counter parameter. - Add optional cancellation and progress support to LogQuery. - Rename TimeBudgetExceeded error to Cancelled. Journal-viewer-plugin (CatalogFunction): - Adapt on_call to receive FunctionCallContext; remove on_cancellation and on_progress implementations. - Set total progress to 2×file count (indexing + querying) and pass the done counter to both phases. - Wrap log querying in spawn_blocking with cancellation support. - Simplify Transaction and TransactionRegistry, removing timeout tracking, progress flags, and cancellation state that the runtime now manages. Agent (pluginsd_functions.c): - Send FUNCTION_CANCEL to the plugin when garbage-collecting timed-out transactions, so the plugin can stop in-progress work. Remove the foundation crate (Timeout is no longer needed). * Be less verbose at info-level. * Add /run/log/journal to default paths. * Move the keep-alive message to the plugin runtime. * Include journal-viewer.yaml in the plugin-journal-viewer RPM subpackage. The config file was excluded from the main netdata package but never added to the plugin-journal-viewer %files section, so it was missing from installed systems.

vkalintiris committed Feb 8, 2026 at 20:51 UTC 09cbde151aaafa5d15b9340d10e656aceb4d445f

29 files changed +551 -601

netdata.spec.in

@@ -3137,6 +3137,7 @@ Requires(pre): %{name}-user >= %{version}

3137 %defattr(0750,root,netdata,0750)

3138 # CAP_DAC_READ_SEARCH required for reading journal files.

3139 %caps(cap_dac_read_search=ep) %attr(0750,root,netdata) %{_libexecdir}/%{name}/plugins.d/journal-viewer-plugin

3140 +%attr(0640,root,netdata) %{_libdir}/%{name}/conf.d/journal-viewer.yaml

3141

3142 %if 0%{?centos_ver} != 7 && 0%{?amazon_linux} != 2

3143 %package plugin-systemd-units

src/crates/Cargo.lock

+2 -9

@@ -855,13 +855,6 @@ dependencies = [

855 "percent-encoding",

856 ]

857

858 -[[package]]

859 -name = "foundation"

860 -version = "0.1.3"

861 -dependencies = [

862 - "tokio",

863 -]

864 -

858 [[package]]

859 name = "foyer"

860 version = "0.20.1"

@@ -1623,7 +1616,6 @@ dependencies = [

1616 "async-stream",

1617 "async-trait",

1618 "chrono",

1626 - "foundation",

1619 "foyer",

1620 "futures",

1621 "journal-common",

@@ -1639,6 +1631,7 @@ dependencies = [

1631 "tempfile",

1632 "thiserror",

1633 "tokio",

1634 + "tokio-util",

1635 "tracing",

1636 "tracing-subscriber",

1637 "uuid",

@@ -1750,6 +1743,7 @@ dependencies = [

1743 "serde_yaml",

1744 "thiserror",

1745 "tokio",

1746 + "tokio-util",

1747 "tracing",

1748 ]

1749

@@ -2740,7 +2734,6 @@ dependencies = [

2734 "async-trait",

2735 "bytes",

2736 "console-subscriber",

2743 - "foundation",

2737 "futures",

2738 "itoa",

2739 "netdata-plugin-charts-derive",

src/crates/Cargo.toml

-2

@@ -11,7 +11,6 @@ members = [

11 "rdp",

13 # Netdata plugin workspace members

14 - "netdata-plugin/foundation",

14 "netdata-plugin/error",

15 "netdata-plugin/protocol",

16 "netdata-plugin/rt",

@@ -136,7 +135,6 @@ journalctl = { path = "journalctl" }

135 rdp = { path = "rdp" }

136

137 # Netdata plugin crates

139 -foundation = { path = "netdata-plugin/foundation" }

138 netdata-plugin-error = { path = "netdata-plugin/error" }

139 netdata-plugin-protocol = { path = "netdata-plugin/protocol" }

140 netdata-plugin-rt = { path = "netdata-plugin/rt" }

src/crates/journal-engine/Cargo.toml

+1 -1

@@ -28,7 +28,7 @@ rayon = { workspace = true }

28 journal-core = { workspace = true }

29 journal-index = { workspace = true }

30 journal-registry = { workspace = true }

31 -foundation = { workspace = true }

31 +tokio-util = { workspace = true }

32 lru = { workspace = true }

34 allocative = { workspace = true, optional = true }

src/crates/journal-engine/examples/index.rs

+5 -6

@@ -30,7 +30,6 @@

30 // # Let's say it's 259:0, Set a 10MB/s read and write limit

31 // echo "259:0 rbps=10485760 wbps=10485760" | sudo tee /sys/fs/cgroup/slow-io/io.max

33 -use foundation::Timeout;

33 use journal_engine::{

34 Facets, FileIndexCacheBuilder, FileIndexKey, IndexingLimits, QueryTimeRange,

35 batch_compute_file_indexes,

@@ -39,7 +38,7 @@ use journal_index::FieldName;

38 use journal_registry::{Monitor, Registry};

39 use std::env;

40 use std::path::PathBuf;

42 -use std::time::Duration;

41 +use tokio_util::sync::CancellationToken;

43 #[allow(unused_imports)]

44 use tracing::{info, warn};

@@ -110,12 +109,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {

109 .duration_since(std::time::UNIX_EPOCH)?

110 .as_secs() as u32;

111 let time_range = QueryTimeRange::new(now - 86400, now)?;

113 - let timeout = Timeout::new(Duration::from_secs(60));

112 + let cancellation = CancellationToken::new();

113

114 info!(

116 - "computing {} file indexes with timeout {:?}, bucket duration: {}s",

115 + "computing {} file indexes, bucket duration: {}s",

116 keys.len(),

118 - timeout.remaining(),

117 time_range.bucket_duration()

118 );

119

@@ -126,8 +124,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {

124 &registry,

125 keys,

126 &time_range,

129 - timeout,

127 + cancellation,

128 IndexingLimits::default(),

129 + None,

130 )

131 .await?;

132

src/crates/journal-engine/src/error.rs

+3 -3

@@ -46,9 +46,9 @@ pub enum EngineError {

46 #[error("Foyer IO error: {0}")]

47 FoyerIo(#[from] foyer::IoError),

49 - /// Time budget exceeded during batch processing

50 - #[error("Time budget exceeded")]

51 - TimeBudgetExceeded,

49 + /// Operation was cancelled

50 + #[error("Operation cancelled")]

51 + Cancelled,

53 /// Invalid time range (start >= end)

54 #[error("Invalid time range: start={start} >= end={end}")]

src/crates/journal-engine/src/indexing.rs

+42 -31

@@ -9,9 +9,11 @@ use crate::{

9 error::{EngineError, Result},

10 query_time_range::QueryTimeRange,

11 };

12 -use foundation::Timeout;

12 +use tokio_util::sync::CancellationToken;

13 use journal_index::{FileIndex, FileIndexer, IndexingLimits};

14 use journal_registry::Registry;

15 +use std::sync::Arc;

16 +use std::sync::atomic::AtomicUsize;

17 use tracing::{error, trace};

19 // ============================================================================

@@ -137,19 +139,21 @@ impl Default for FileIndexCacheBuilder {

139 /// * `registry` - Registry to update with file metadata

140 /// * `keys` - Vector of (file, facets, source_timestamp_field) to fetch/compute indexes for

141 /// * `time_range` - Query time range for bucket duration calculation

140 -/// * `timeout` - Timeout for the entire operation (can be extended dynamically)

142 +/// * `cancellation` - Token to signal cancellation from the caller

143 /// * `indexing_limits` - Configuration limits for indexing (cardinality, payload size)

144 +/// * `progress_counter` - Optional atomic counter incremented after each file is indexed

145 ///

146 /// # Returns

147 /// Vector of responses for each key. Successful responses contain the file index.

145 -/// If timeout expires, returns TimeBudgetExceeded error.

148 +/// If cancelled, returns Cancelled error.

149 pub async fn batch_compute_file_indexes(

150 cache: &FileIndexCache,

151 registry: &Registry,

152 keys: Vec<FileIndexKey>,

153 time_range: &QueryTimeRange,

151 - timeout: Timeout,

154 + cancellation: CancellationToken,

155 indexing_limits: IndexingLimits,

156 + progress_counter: Option<Arc<AtomicUsize>>,

157 ) -> Result<Vec<(FileIndexKey, FileIndex)>> {

158 let bucket_duration = time_range.bucket_duration_seconds();

159 // Phase 1: Batch check cache for all keys upfront

@@ -165,13 +169,10 @@ pub async fn batch_compute_file_indexes(

169 }

170 });

171

168 - let cache_lookup_results: Vec<(FileIndexKey, Result<Option<FileIndex>>)> =

169 - tokio::time::timeout(

170 - timeout.remaining(),

171 - futures::future::join_all(cache_lookup_futures),

172 - )

173 - .await

174 - .map_err(|_| EngineError::TimeBudgetExceeded)?;

172 + let cache_lookup_results: Vec<(FileIndexKey, Result<Option<FileIndex>>)> = tokio::select! {

173 + results = futures::future::join_all(cache_lookup_futures) => results,

174 + _ = cancellation.cancelled() => return Err(EngineError::Cancelled),

175 + };

176

177 // Phase 2: Separate cache hits from misses, check freshness and compatibility

178 let mut responses = Vec::with_capacity(keys.len());

@@ -213,8 +214,8 @@ pub async fn batch_compute_file_indexes(

214 }

215 }

216

216 - if timeout.is_expired() {

217 - return Err(EngineError::TimeBudgetExceeded);

217 + if cancellation.is_cancelled() {

218 + return Err(EngineError::Cancelled);

219 }

220

221 trace!(

@@ -223,23 +224,25 @@ pub async fn batch_compute_file_indexes(

224 );

225

226 // Phase 3: Spawn single blocking task with rayon for parallel computation

226 - let time_budget_remaining = timeout.remaining();

227 + //

228 + // The cancellation token is cloned into the blocking task so that cancellation

229 + // is visible to the per-file check.

230 + let cancellation_for_blocking = cancellation.clone();

231

232 let compute_task = tokio::task::spawn_blocking(move || {

233 use rayon::prelude::*;

234 use std::sync::Arc;

235 use std::sync::atomic::{AtomicBool, Ordering};

236

233 - let deadline = std::time::Instant::now() + time_budget_remaining;

234 - let timed_out = Arc::new(AtomicBool::new(false));

237 + let cancelled = Arc::new(AtomicBool::new(false));

238

239 keys_to_compute

240 .into_par_iter()

241 .map(|key| {

239 - // Check time budget before processing

240 - if std::time::Instant::now() >= deadline || timed_out.load(Ordering::Relaxed) {

241 - timed_out.store(true, Ordering::Relaxed);

242 - return (key, Err(EngineError::TimeBudgetExceeded));

242 + // Check cancellation before processing

243 + if cancellation_for_blocking.is_cancelled() || cancelled.load(Ordering::Relaxed) {

244 + cancelled.store(true, Ordering::Relaxed);

245 + return (key, Err(EngineError::Cancelled));

246 }

247

248 let mut file_indexer = FileIndexer::new(indexing_limits);

@@ -252,23 +255,31 @@ pub async fn batch_compute_file_indexes(

255 )

256 .map_err(|e| e.into());

257

258 + if result.is_ok() {

259 + if let Some(ref counter) = progress_counter {

260 + counter.fetch_add(1, Ordering::Relaxed);

261 + }

262 + }

263 +

264 (key, result)

265 })

266 .collect::<Vec<(FileIndexKey, Result<FileIndex>)>>()

267 });

268

260 - let computed_results = match tokio::time::timeout(time_budget_remaining, compute_task).await {

261 - Ok(Ok(results)) => results,

262 - Ok(Err(e)) => {

263 - return Err(EngineError::Io(std::io::Error::new(

264 - std::io::ErrorKind::Other,

265 - format!("Blocking task panicked: {}", e),

266 - )));

269 + let computed_results = tokio::select! {

270 + result = compute_task => {

271 + match result {

272 + Ok(results) => results,

273 + Err(e) => {

274 + return Err(EngineError::Io(std::io::Error::new(

275 + std::io::ErrorKind::Other,

276 + format!("Blocking task panicked: {}", e),

277 + )));

278 + }

279 + }

280 }

268 - Err(_timeout) => {

269 - // Note: the blocking task will continue running in background but

270 - // we will ignore the results

271 - return Err(EngineError::TimeBudgetExceeded);

281 + _ = cancellation.cancelled() => {

282 + return Err(EngineError::Cancelled);

283 }

284 };

285

src/crates/journal-engine/src/logs/query.rs

+64 -11

@@ -13,6 +13,9 @@ use journal_index::{

13 use journal_registry::File;

14 use std::collections::HashMap;

15 use std::num::NonZeroU64;

16 +use std::sync::Arc;

17 +use std::sync::atomic::{AtomicUsize, Ordering};

18 +use tokio_util::sync::CancellationToken;

19 use tracing::warn;

21 /// Pagination state for multi-file log queries.

@@ -51,6 +54,8 @@ pub struct PaginationState {

54 pub struct LogQuery<'a> {

55 file_indexes: &'a [FileIndex],

56 builder: LogQueryParamsBuilder,

57 + cancellation: Option<CancellationToken>,

58 + progress: Option<Arc<AtomicUsize>>,

59 }

61 impl<'a> LogQuery<'a> {

@@ -74,6 +79,8 @@ impl<'a> LogQuery<'a> {

79 builder: LogQueryParamsBuilder::new(anchor, direction).with_source_timestamp_field(

80 Some(FieldName::new_unchecked("_SOURCE_REALTIME_TIMESTAMP")),

81 ),

82 + cancellation: None,

83 + progress: None,

84 }

85 }

@@ -132,6 +139,24 @@ impl<'a> LogQuery<'a> {

139 self

140 }

141

142 + /// Set a cancellation token for the query (optional).

143 + ///

144 + /// When set, the query will check the token before processing each file

145 + /// and return early with partial results if cancelled.

146 + pub fn with_cancellation(mut self, token: CancellationToken) -> Self {

147 + self.cancellation = Some(token);

148 + self

149 + }

150 +

151 + /// Set a progress counter for the query (optional).

152 + ///

153 + /// When set, the counter is incremented (via `fetch_add`) after each file

154 + /// is processed in `retrieve_log_entries`.

155 + pub fn with_progress(mut self, counter: Arc<AtomicUsize>) -> Self {

156 + self.progress = Some(counter);

157 + self

158 + }

159 +

160 /// Execute the query and return log entries.

161 ///

162 /// This consumes the builder and returns a vector of log entries sorted by timestamp

@@ -142,8 +167,13 @@ impl<'a> LogQuery<'a> {

167 /// Returns an error if anchor or direction were not set, or if time boundaries are invalid.

168 pub fn execute(self) -> Result<Vec<LogEntryData>> {

169 let params = self.builder.build()?;

145 - let (log_entry_ids, _state) =

146 - retrieve_log_entries(self.file_indexes.to_vec(), params, None);

170 + let (log_entry_ids, _state) = retrieve_log_entries(

171 + self.file_indexes.to_vec(),

172 + params,

173 + None,

174 + self.cancellation.as_ref(),

175 + self.progress.as_ref(),

176 + );

177

178 extract_entry_data(&log_entry_ids)

179 }

@@ -170,8 +200,13 @@ impl<'a> LogQuery<'a> {

200 state: Option<&PaginationState>,

201 ) -> Result<(Vec<LogEntryData>, PaginationState)> {

202 let params = self.builder.build()?;

173 - let (log_entry_ids, new_state) =

174 - retrieve_log_entries(self.file_indexes.to_vec(), params, state);

203 + let (log_entry_ids, new_state) = retrieve_log_entries(

204 + self.file_indexes.to_vec(),

205 + params,

206 + state,

207 + self.cancellation.as_ref(),

208 + self.progress.as_ref(),

209 + );

210

211 let data = extract_entry_data(&log_entry_ids)?;

212 Ok((data, new_state))

@@ -197,6 +232,8 @@ fn retrieve_log_entries(

232 file_indexes: Vec<FileIndex>,

233 params: LogQueryParams,

234 state: Option<&PaginationState>,

235 + cancellation: Option<&CancellationToken>,

236 + progress: Option<&Arc<AtomicUsize>>,

237 ) -> (Vec<LogEntryId>, PaginationState) {

238 // Handle edge cases

239 if params.limit() == Some(0) || file_indexes.is_empty() {

@@ -242,6 +279,11 @@ fn retrieve_log_entries(

279 }

280 };

281

282 + if let Some(counter) = progress {

283 + let filtered = file_indexes.len() - relevant_indexes.len();

284 + counter.fetch_add(filtered, Ordering::Relaxed);

285 + }

286 +

287 if relevant_indexes.is_empty() {

288 return (Vec::new(), PaginationState::default());

289 }

@@ -268,6 +310,21 @@ fn retrieve_log_entries(

310 let mut new_state = state.cloned().unwrap_or_default();

311

312 for file_index in relevant_indexes {

313 + // Check cancellation before processing each file

314 + if let Some(token) = cancellation {

315 + if token.is_cancelled() {

316 + warn!(

317 + "log query cancelled after processing {} files, returning partial results",

318 + new_state.file_positions.len()

319 + );

320 + break;

321 + }

322 + }

323 +

324 + if let Some(counter) = progress {

325 + counter.fetch_add(1, Ordering::Relaxed);

326 + }

327 +

328 // Pruning optimization: if we have a full result set, check if we can skip

329 // remaining files based on their time ranges

330 if collected_entries.len() >= limit {

@@ -321,14 +378,10 @@ fn retrieve_log_entries(

378 }

379 };

380

324 - if new_entries.is_empty() {

325 - continue;

381 + if !new_entries.is_empty() {

382 + collected_entries =

383 + merge_log_entries(collected_entries, new_entries, limit, params.direction());

384 }

327 -

328 - // Merge the new entries with our existing results, maintaining

329 - // sorted order and respecting the limit constraint

330 - collected_entries =

331 - merge_log_entries(collected_entries, new_entries, limit, params.direction());

385 }

386

387 // Update pagination state based on the last position for each file in collected_entries

src/crates/journal-index/src/file_index.rs

+5 -5

@@ -7,7 +7,7 @@ use journal_core::repository::File;

7 use regex::Regex;

8 use serde::{Deserialize, Serialize};

9 use std::num::NonZeroU64;

10 -use tracing::{debug, error};

10 +use tracing::{error, trace};

12 /// Index for a single journal file, enabling efficient querying and filtering.

13 ///

@@ -348,10 +348,10 @@ impl LogQueryParamsBuilder {

348

349 // Compile regex pattern if provided

350 let regex = if let Some(pattern) = self.regex_pattern {

351 - debug!("compiling regex pattern for log query: {:?}", pattern);

351 + trace!("compiling regex pattern for log query: {:?}", pattern);

352 match Regex::new(&pattern) {

353 Ok(regex) => {

354 - debug!("regex pattern compiled successfully");

354 + trace!("regex pattern compiled successfully");

355 Some(regex)

356 }

357 Err(e) => {

@@ -599,7 +599,7 @@ impl FileIndex {

599 // Log if regex filtering is active

600 let mut regex_filtered_count = 0usize;

601 if params.regex().is_some() {

602 - debug!(

602 + trace!(

603 "regex filtering enabled for query, will filter {} candidate entries",

604 entry_offsets.len()

605 );

@@ -795,7 +795,7 @@ impl FileIndex {

795

796 // Log regex filtering statistics if regex was used

797 if params.regex().is_some() {

798 - debug!(

798 + trace!(

799 "regex filtering complete: {} entries matched, {} entries filtered out",

800 log_entry_ids.len(),

801 regex_filtered_count

src/crates/journal-registry/src/registry/mod.rs

+7 -8

@@ -17,7 +17,7 @@ use notify::{

17 };

18 use parking_lot::RwLock;

19 use std::sync::Arc;

20 -use tracing::{debug, error, info, trace, warn};

20 +use tracing::{error, info, trace, warn};

22 mod monitor;

23 pub use monitor::Monitor;

@@ -186,7 +186,7 @@ impl Registry {

186

187 // Insert all discovered files into repository (automatically initializes metadata)

188 for file in files {

189 - debug!("adding file to repository: {:?}", file.path());

189 + trace!("adding file to repository: {:?}", file.path());

190

191 if let Err(e) = inner.repository.insert(file) {

192 error!("failed to insert file into repository: {}", e);

@@ -227,7 +227,7 @@ impl Registry {

227 match event.kind {

228 EventKind::Create(_) => {

229 for path in &event.paths {

230 - debug!("adding file to repository: {:?}", path);

230 + trace!("adding file to repository: {:?}", path);

231

232 if let Some(file) = File::from_path(path) {

233 if let Err(e) = inner.repository.insert(file) {

@@ -240,7 +240,7 @@ impl Registry {

240 }

241 EventKind::Remove(_) => {

242 for path in &event.paths {

243 - debug!("removing file from repository: {:?}", path);

243 + trace!("removing file from repository: {:?}", path);

244

245 if let Some(file) = File::from_path(path) {

246 if let Err(e) = inner.repository.remove(&file) {

@@ -284,10 +284,9 @@ impl Registry {

284 rename_mode

285 );

286 }

287 - event_kind => {

288 - // Ignore other events (content modifications, access, etc.)

289 - trace!("ignoring notify event kind: {:?}", event_kind);

290 - }

287 +

288 + // Ignore other events (content modifications, access, etc.)

289 + _ => {}

290 }

291 Ok(())

292 }

src/crates/netdata-log-viewer/journal-function/src/lib.rs

-3

@@ -16,9 +16,6 @@ pub use journal_engine::{

16 calculate_bucket_duration, entry_data_to_table,

17 };

19 -// Re-export Timeout from foundation (via rt for backward compatibility)

20 -pub use rt::Timeout;

21 -

19 // Re-export Netdata-specific charts/metrics

20 pub use charts::{

21 BucketCacheMetrics, BucketOperationsMetrics, FileIndexingMetrics, JournalMetrics,

src/crates/netdata-log-viewer/journal-viewer-plugin/Cargo.toml

@@ -23,6 +23,7 @@ serde = { workspace = true }

23 serde_json = { workspace = true }

24 tracing = { workspace = true}

25 tokio = { workspace = true }

26 +tokio-util = { workspace = true }

27 notify = { workspace = true }

28 thiserror = { workspace = true }

29 foyer = { workspace = true }

src/crates/netdata-log-viewer/journal-viewer-plugin/src/catalog.rs

+96 -145

@@ -5,9 +5,11 @@ use netdata_plugin_error::Result;

5 use netdata_plugin_protocol::FunctionDeclaration;

6 use netdata_plugin_schema::HttpAccess;

7 use parking_lot::RwLock;

8 -use rt::FunctionHandler;

8 +use rt::{FunctionCallContext, FunctionHandler};

9 use std::sync::Arc;

10 -use tracing::{debug, error, info, instrument, warn};

10 +use std::sync::atomic::AtomicUsize;

11 +use tokio_util::sync::CancellationToken;

12 +use tracing::{error, instrument, trace, warn};

14 // Import types from journal-function crate

15 use journal_function::{

@@ -98,15 +100,11 @@ fn required_params() -> Vec<netdata::RequiredParam> {

100 struct TransactionInner {

101 id: String,

102 start_time: tokio::time::Instant,

101 - report_progress: bool,

102 - cancel_call: bool,

103 - timeout: Option<journal_function::Timeout>,

103 }

104

105 /// Represents a tracked transaction for a function call.

106 ///

108 -/// Transactions track the lifecycle and state of individual function calls,

109 -/// allowing for cancellation checks, progress reporting, and timeout detection.

107 +/// Transactions track the lifecycle of individual function calls.

108 #[derive(Debug, Clone)]

109 struct Transaction {

110 inner: Arc<RwLock<TransactionInner>>,

@@ -114,14 +112,11 @@ struct Transaction {

112

113 impl Transaction {

114 /// Create a new transaction with the given ID.

117 - fn new(id: String, timeout: Option<journal_function::Timeout>) -> Self {

115 + fn new(id: String) -> Self {

116 Self {

117 inner: Arc::new(RwLock::new(TransactionInner {

118 id,

119 start_time: tokio::time::Instant::now(),

122 - report_progress: false,

123 - cancel_call: false,

124 - timeout,

120 })),

121 }

122 }

@@ -131,41 +126,10 @@ impl Transaction {

126 self.inner.read().id.clone()

127 }

128

134 - /// Check if the transaction has been marked for cancellation.

135 - #[allow(dead_code)]

136 - fn is_cancelled(&self) -> bool {

137 - self.inner.read().cancel_call

138 - }

139 -

140 - /// Mark the transaction for cancellation.

141 - fn cancel(&self) {

142 - self.inner.write().cancel_call = true;

143 - }

144 -

145 - /// Check if progress reporting is requested for this transaction.

146 - #[allow(dead_code)]

147 - fn should_report_progress(&self) -> bool {

148 - self.inner.read().report_progress

149 - }

150 -

151 - /// Set the progress reporting flag.

152 - fn set_report_progress(&self, report: bool) {

153 - self.inner.write().report_progress = report;

154 - }

155 -

129 /// Get the elapsed time since the transaction started.

130 fn elapsed(&self) -> std::time::Duration {

131 self.inner.read().start_time.elapsed()

132 }

160 -

161 - /// Reset the timeout to the initial budget from the current time.

162 - ///

163 - /// This is called when progress is reported to give the operation its full timeout budget again.

164 - fn reset_timeout(&self) {

165 - if let Some(timeout) = &self.inner.read().timeout {

166 - timeout.reset();

167 - }

168 - }

133 }

134

135 /// Registry for managing active transactions.

@@ -188,11 +152,7 @@ impl TransactionRegistry {

152 /// Create and register a new transaction with the given ID.

153 ///

154 /// Returns None if a transaction with this ID already exists.

191 - fn create(

192 - &self,

193 - id: String,

194 - timeout: Option<journal_function::Timeout>,

195 - ) -> Option<Transaction> {

155 + fn create(&self, id: String) -> Option<Transaction> {

156 let mut transactions = self.transactions.write();

157

158 if transactions.contains_key(&id) {

@@ -200,49 +160,17 @@ impl TransactionRegistry {

160 return None;

161 }

162

203 - let transaction = Transaction::new(id.clone(), timeout);

163 + let transaction = Transaction::new(id.clone());

164 transactions.insert(id, transaction.clone());

165

166 Some(transaction)

167 }

168

209 - /// Get an existing transaction by ID.

210 - fn get(&self, id: &str) -> Option<Transaction> {

211 - self.transactions.read().get(id).cloned()

212 - }

213 -

169 /// Remove a transaction from the registry.

170 ///

171 /// Returns the removed transaction if it existed.

172 fn remove(&self, id: &str) -> Option<Transaction> {

218 - let transaction = self.transactions.write().remove(id);

219 - transaction

220 - }

221 -

222 - /// Cancel a transaction by ID.

223 - ///

224 - /// Returns true if the transaction was found and cancelled.

225 - fn cancel(&self, id: &str) -> bool {

226 - if let Some(transaction) = self.get(id) {

227 - transaction.cancel();

228 - info!("Cancelled transaction {}", id);

229 - true

230 - } else {

231 - warn!("Cannot cancel non-existent transaction {}", id);

232 - false

233 - }

234 - }

235 -

236 - /// Get the number of active transactions.

237 - #[allow(dead_code)]

238 - fn len(&self) -> usize {

239 - self.transactions.read().len()

240 - }

241 -

242 - /// Check if the registry is empty.

243 - #[allow(dead_code)]

244 - fn is_empty(&self) -> bool {

245 - self.transactions.read().is_empty()

173 + self.transactions.write().remove(id)

174 }

175 }

176

@@ -273,7 +201,6 @@ impl CatalogFunction {

201 /// - has_before: true if there are more entries before the returned window

202 /// - has_after: true if there are more entries after the returned window

203 fn query_logs_from_indexes(

276 - &self,

204 indexed_files: &[journal_index::FileIndex],

205 time_range: &journal_function::QueryTimeRange,

206 anchor: Option<u64>,

@@ -281,6 +208,8 @@ impl CatalogFunction {

208 search_query: &str,

209 limit: usize,

210 direction: journal_index::Direction,

211 + cancellation: Option<CancellationToken>,

212 + progress: Option<Arc<AtomicUsize>>,

213 ) -> (Vec<journal_function::LogEntryData>, bool, bool) {

214 use journal_function::LogQuery;

215

@@ -325,6 +254,14 @@ impl CatalogFunction {

254 .with_after_usec(after_usec)

255 .with_before_usec(before_usec);

256

257 + if let Some(ref t) = cancellation {

258 + query = query.with_cancellation(t.clone());

259 + }

260 +

261 + if let Some(counter) = progress {

262 + query = query.with_progress(counter);

263 + }

264 +

265 // Only apply filter if it's not Filter::none() (which matches nothing)

266 if !filter.is_none() {

267 query = query.with_filter(filter.clone());

@@ -382,6 +319,10 @@ impl CatalogFunction {

319 .with_after_usec(after_usec)

320 .with_before_usec(before_usec);

321

322 + if let Some(ref t) = cancellation {

323 + opposite_query = opposite_query.with_cancellation(t.clone());

324 + }

325 +

326 // Only apply filter if it's not Filter::none() (which matches nothing)

327 if !filter.is_none() {

328 opposite_query = opposite_query.with_filter(filter.clone());

@@ -389,7 +330,7 @@ impl CatalogFunction {

330

331 // Apply regex search if search_query is not empty

332 if !search_query.is_empty() {

392 - debug!("applying regex filter to opposite direction query");

333 + trace!("applying regex filter to opposite direction query");

334 opposite_query = opposite_query.with_regex(search_query);

335 }

336

@@ -490,19 +431,24 @@ impl FunctionHandler for CatalogFunction {

431 type Request = CatalogRequest;

432 type Response = CatalogResponse;

433

493 - async fn on_call(&self, transaction: String, request: Self::Request) -> Result<Self::Response> {

494 - // Register the transaction with the timeout

495 - let timeout = journal_function::Timeout::new(std::time::Duration::from_secs(10));

434 + async fn on_call(

435 + &self,

436 + ctx: FunctionCallContext,

437 + request: Self::Request,

438 + ) -> Result<Self::Response> {

439 + let transaction = ctx.transaction();

440 +

441 + // Register the transaction

442 let Some(txn) = self

443 .inner

444 .transaction_registry

499 - .create(transaction.clone(), Some(timeout.clone()))

445 + .create(transaction.to_owned())

446 else {

447 return Err(netdata_plugin_error::NetdataPluginError::Other {

448 message: format!("[{}] transaction already exists", transaction),

449 });

450 };

505 - debug!("[{}] started transaction", txn.id());

451 + trace!("[{}] started transaction", txn.id());

452

453 // Create query time range with automatic alignment

454 let time_range = journal_function::QueryTimeRange::new(request.after, request.before)

@@ -511,7 +457,7 @@ impl FunctionHandler for CatalogFunction {

457 netdata_plugin_error::NetdataPluginError::Other { message: msg }

458 })?;

459

514 - debug!(

460 + trace!(

461 "[{}] time range: [{}, {}), aligned: [{}, {}), bucket duration: {} seconds",

462 txn.id(),

463 time_range.requested_start(),

@@ -532,7 +478,7 @@ impl FunctionHandler for CatalogFunction {

478 netdata_plugin_error::NetdataPluginError::Other { message: msg }

479 })?;

480 let find_files_duration = op_start.elapsed();

535 - debug!("[{}] found {} files in time range", txn.id(), files.len(),);

481 + trace!("[{}] found {} files in time range", txn.id(), files.len(),);

482 if tracing::enabled!(tracing::Level::TRACE) {

483 for (idx, file_info) in files.iter().enumerate() {

484 tracing::trace!(

@@ -547,11 +493,11 @@ impl FunctionHandler for CatalogFunction {

493

494 // Build filter expression

495 let filter_expr = build_filter_from_selections(&request.selections);

550 - debug!("[{}] filter expression: {}", txn.id(), filter_expr);

496 + trace!("[{}] filter expression: {}", txn.id(), filter_expr);

497

498 // Build facets for file indexes

499 let facets = Facets::new(&request.facets);

554 - debug!(

500 + trace!(

501 "[{}] using {} facets with precomputed hash {}",

502 txn.id(),

503 facets.len(),

@@ -565,15 +511,23 @@ impl FunctionHandler for CatalogFunction {

511 .map(|f| FileIndexKey::new(&f.file, &facets, Some(source_timestamp_field.clone())))

512 .collect();

513

568 - // Index all files

514 + // Progress is reported in two phases: indexing and querying.

515 + // Start with total = number of files for the indexing phase. After

516 + // indexing completes we extend the total so the query phase gets its

517 + // own progress range. This avoids over-estimating total work when the

518 + // second phase is fast (which is the common case).

519 + let num_files = keys.len();

520 + ctx.progress.set_total(num_files);

521 +

522 let op_start = std::time::Instant::now();

523 let indexed_files = journal_function::batch_compute_file_indexes(

524 &self.inner.cache,

525 &self.inner.registry,

526 keys,

527 &time_range,

575 - timeout,

528 + ctx.cancellation.clone(),

529 self.inner.indexing_limits,

530 + Some(ctx.progress.done_counter()),

531 )

532 .await

533 .map_err(|e| {

@@ -582,7 +536,12 @@ impl FunctionHandler for CatalogFunction {

536 })?;

537 let indexing_duration = op_start.elapsed();

538

585 - debug!(

539 + // Extend progress to cover the query phase. The done counter is

540 + // already at ~keys.len(), so the UI will show ~50% until querying

541 + // catches up.

542 + ctx.progress.set_total(2 * num_files);

543 +

544 + trace!(

545 "[{}] retrieved {}/{} file indexes for histogram buckets and log entries",

546 txn.id(),

547 indexed_files.len(),

@@ -615,21 +574,51 @@ impl FunctionHandler for CatalogFunction {

574 })?;

575 let histogram_duration = op_start.elapsed();

576

618 - // Query logs from pre-indexed files

577 + // Query logs from pre-indexed files, wrapped in spawn_blocking so the

578 + // async runtime can handle cancellation while this runs.

579 let op_start = std::time::Instant::now();

580 let limit = request.last.unwrap_or(200);

581 let file_indexes: Vec<_> = indexed_files.iter().map(|(_, idx)| idx.clone()).collect();

622 - let (log_entries, has_before, has_after) = self.query_logs_from_indexes(

623 - &file_indexes,

624 - &time_range,

625 - request.anchor,

626 - &filter_expr,

627 - &request.query,

628 - limit,

629 - request.direction,

630 - );

582 + let query_progress = ctx.progress.done_counter();

583 +

584 + let query_filter = filter_expr.clone();

585 + let query_search = request.query.clone();

586 + let query_direction = request.direction;

587 + let query_anchor = request.anchor;

588 + let query_time_range = time_range.clone();

589 + let query_cancellation = ctx.cancellation.clone();

590 +

591 + let query_task = tokio::task::spawn_blocking(move || {

592 + CatalogFunction::query_logs_from_indexes(

593 + &file_indexes,

594 + &query_time_range,

595 + query_anchor,

596 + &query_filter,

597 + &query_search,

598 + limit,

599 + query_direction,

600 + Some(query_cancellation),

601 + Some(query_progress),

602 + )

603 + });

604 +

605 + let (log_entries, has_before, has_after) = tokio::select! {

606 + result = query_task => {

607 + match result {

608 + Ok(result) => result,

609 + Err(e) => {

610 + error!("[{}] log query task panicked: {}", txn.id(), e);

611 + (Vec::new(), false, false)

612 + }

613 + }

614 + }

615 + _ = ctx.cancellation.cancelled() => {

616 + warn!("[{}] log query cancelled", txn.id());

617 + (Vec::new(), false, false)

618 + }

619 + };

620 let query_logs_duration = op_start.elapsed();

632 - debug!(

621 + trace!(

622 "[{}] retrieved {} log entries (has before: {}, has after: {})",

623 txn.id(),

624 log_entries.len(),

@@ -690,7 +679,7 @@ impl FunctionHandler for CatalogFunction {

679 message: format!("[{}] transaction does not exist", transaction),

680 });

681 };

693 - debug!(

682 + trace!(

683 "[{}] completed transaction (find_files: {:?}, indexing: {:?}, histogram: {:?}, query_logs: {:?}, total: {:?})",

684 txn.id(),

685 find_files_duration,

@@ -703,49 +692,11 @@ impl FunctionHandler for CatalogFunction {

692 Ok(response)

693 }

694

706 - async fn on_cancellation(&self, transaction: String) -> Result<Self::Response> {

707 - warn!("catalog function call {} cancelled by Netdata", transaction);

708 -

709 - // Mark the transaction as cancelled

710 - self.inner.transaction_registry.cancel(&transaction);

711 -

712 - // Remove the transaction from the registry

713 - self.inner.transaction_registry.remove(&transaction);

714 -

715 - Err(netdata_plugin_error::NetdataPluginError::Other {

716 - message: "catalog function cancelled by user".to_string(),

717 - })

718 - }

719 -

720 - async fn on_progress(&self, transaction: String) {

721 - info!(

722 - "progress report requested for catalog function call {}",

723 - transaction

724 - );

725 -

726 - // Mark the transaction for progress reporting and reset the timeout

727 - if let Some(txn) = self.inner.transaction_registry.get(&transaction) {

728 - txn.set_report_progress(true);

729 - txn.reset_timeout();

730 - info!(

731 - "Transaction {} marked for progress reporting and timeout reset to initial budget (elapsed: {:?})",

732 - transaction,

733 - txn.elapsed()

734 - );

735 - } else {

736 - warn!(

737 - "Progress requested for non-existent transaction {}",

738 - transaction

739 - );

740 - }

741 - }

742 -

695 fn declaration(&self) -> FunctionDeclaration {

696 // NOTE: `rt` special cases this function call to handle GET/POST

697 // calls in a consistent way. If you rename this function, you should

698 // update the `rt` crate as well.

699

748 - info!("generating journal-viewer function declaration");

700 let mut func_decl = FunctionDeclaration::new(

701 "journal-viewer",

702 "Query and visualize journal log entries with histograms and facets",

src/crates/netdata-log-viewer/journal-viewer-plugin/src/main.rs

+3 -33

@@ -66,7 +66,6 @@ async fn run_plugin() -> std::result::Result<(), Box<dyn std::error::Error>> {

66 };

68 // Create catalog function with disk-backed cache

69 - info!("creating catalog function with Foyer hybrid cache");

69 let indexing_limits = IndexingLimits {

70 max_unique_values_per_field: config.indexing.max_unique_values_per_field,

71 max_field_payload_size: config.indexing.max_field_payload_size,

@@ -79,22 +78,15 @@ async fn run_plugin() -> std::result::Result<(), Box<dyn std::error::Error>> {

78 indexing_limits,

79 )

80 .await?;

82 - info!("catalog function initialized");

82 // Watch configured journal directories

83 for path in &config.journal.paths {

86 - match catalog_function.watch_directory(path) {

87 - Ok(()) => {

88 - info!("watching journal directory: {}", path);

89 - }

90 - Err(e) => {

91 - error!("failed to watch directory {}: {:#?}", path, e);

92 - }

84 + if let Err(e) = catalog_function.watch_directory(path) {

85 + error!("failed to watch directory {}: {:#?}", path, e);

86 }

87 }

89 runtime.register_handler(catalog_function.clone());

97 - info!("catalog function handler registered");

91 // Spawn task to process notify events

92 let catalog_function_clone = catalog_function.clone();

@@ -106,29 +98,7 @@ async fn run_plugin() -> std::result::Result<(), Box<dyn std::error::Error>> {

98 info!("notify event processing task terminated");

99 });

100

109 - // Keepalive future to prevent Netdata from killing the plugin

110 - let writer = runtime.writer();

111 - let keepalive = async move {

112 - let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60));

113 - loop {

114 - interval.tick().await;

115 - if let Ok(mut w) = writer.try_lock() {

116 - let _ = w.write_raw(b"PLUGIN_KEEPALIVE\n").await;

117 - }

118 - }

119 - };

120 -

121 - info!("starting plugin runtime");

122 -

123 - // Run plugin runtime and keepalive concurrently

124 - tokio::select! {

125 - result = runtime.run() => {

126 - result?;

127 - }

128 - _ = keepalive => {

129 - // Keepalive loop never completes normally

130 - }

131 - }

101 + runtime.run().await?;

102

103 Ok(())

104 }

src/crates/netdata-log-viewer/journal-viewer-plugin/src/plugin_config.rs

+4 -1

@@ -21,7 +21,10 @@ pub struct JournalConfig {

21 impl Default for JournalConfig {

22 fn default() -> Self {

23 Self {

24 - paths: vec![String::from("/var/log/journal")],

24 + paths: vec![

25 + String::from("/var/log/journal"),

26 + String::from("/run/log/journal"),

27 + ],

28 }

29 }

30 }

src/crates/netdata-otel/otel-plugin/src/lib.rs

+1 -16

@@ -122,19 +122,7 @@ async fn run_internal() -> Result<()> {

122 )

123 .serve(addr);

124

125 - // 10. Keepalive future (PluginRuntime doesn't send keepalive automatically)

126 - let writer_clone = writer.clone();

127 - let keepalive = async move {

128 - let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60));

129 - loop {

130 - interval.tick().await;

131 - if let Ok(mut w) = writer_clone.try_lock() {

132 - let _ = w.write_raw(b"PLUGIN_KEEPALIVE\n").await;

133 - }

134 - }

135 - };

136 -

137 - // 11. Run gRPC server, plugin runtime, and keepalive concurrently

125 + // 10. Run gRPC server and plugin runtime concurrently

126 tokio::select! {

127 result = grpc_server => {

128 result.with_context(|| format!("gRPC server error on {}", config.endpoint.path))?;

@@ -142,9 +130,6 @@ async fn run_internal() -> Result<()> {

130 result = runtime.run() => {

131 result.context("PluginRuntime error")?;

132 }

145 - _ = keepalive => {

146 - // Keepalive loop never completes normally

147 - }

133 }

134

135 Ok(())

src/crates/netdata-plugin/foundation/Cargo.toml deleted

-11

@@ -1,11 +0,0 @@

1 -[package]

2 -name = "foundation"

3 -version.workspace = true

4 -edition.workspace = true

5 -rust-version.workspace = true

6 -

7 -[lints]

8 -workspace = true

9 -

10 -[dev-dependencies]

11 -tokio = { workspace = true, features = ["rt", "time", "macros"] }

src/crates/netdata-plugin/foundation/src/lib.rs deleted

-8

@@ -1,8 +0,0 @@

1 -//! Foundational utilities for Netdata plugins.

2 -//!

3 -//! This crate provides low-level primitives and utilities that other plugin

4 -//! crates build upon, including async operation management and control flow.

5 -

6 -// Timeout management

7 -pub mod timeout;

8 -pub use timeout::Timeout;

src/crates/netdata-plugin/foundation/src/timeout.rs deleted

-161

@@ -1,161 +0,0 @@

1 -//! Timeout management for async operations with thread-safe deadline modification.

2 -

3 -use std::sync::Arc;

4 -use std::sync::atomic::{AtomicU64, Ordering};

5 -use std::time::{Duration, Instant};

6 -

7 -/// A thread-safe timeout that can be checked and extended from multiple threads.

8 -///

9 -/// This is useful for operations that need:

10 -/// - Dynamic timeout extension (e.g., on progress updates)

11 -/// - Parallel checks across multiple async tasks

12 -/// - Future support for cancellation signals

13 -///

14 -/// Extensions are bounded: the remaining time will never exceed the initial budget.

15 -#[derive(Debug, Clone)]

16 -pub struct Timeout {

17 - start: Instant,

18 - budget_us: u64,

19 - deadline_us: Arc<AtomicU64>,

20 -}

21 -

22 -impl Timeout {

23 - /// Create a new timeout with the given budget.

24 - pub fn new(budget: Duration) -> Self {

25 - let start = Instant::now();

26 - let budget_us = budget.as_micros() as u64;

27 -

28 - Self {

29 - start,

30 - budget_us,

31 - deadline_us: Arc::new(AtomicU64::new(budget_us)),

32 - }

33 - }

34 -

35 - /// Check if the timeout has expired.

36 - pub fn is_expired(&self) -> bool {

37 - self.remaining().is_zero()

38 - }

39 -

40 - /// Get remaining time. Returns Duration::ZERO if expired.

41 - pub fn remaining(&self) -> Duration {

42 - let deadline_us = self.deadline_us.load(Ordering::Relaxed);

43 - let elapsed_us = self.start.elapsed().as_micros() as u64;

44 -

45 - if elapsed_us >= deadline_us {

46 - Duration::ZERO

47 - } else {

48 - Duration::from_micros(deadline_us - elapsed_us)

49 - }

50 - }

51 -

52 - /// Reset the timeout to the initial budget from the current time.

53 - ///

54 - /// This is useful for operations that report progress and should

55 - /// get their full timeout budget again.

56 - ///

57 - /// For example, if the initial timeout was 10 seconds and we're at t=5s with 5s remaining,

58 - /// calling reset() will give the operation another 10s (deadline becomes t=15s).

59 - pub fn reset(&self) {

60 - let elapsed_us = self.start.elapsed().as_micros() as u64;

61 - let deadline_us = elapsed_us + self.budget_us;

62 - self.deadline_us.store(deadline_us, Ordering::Relaxed);

63 - }

64 -}

65 -

66 -#[cfg(test)]

67 -mod tests {

68 - use super::*;

69 - use std::thread;

70 -

71 - #[test]

72 - fn test_timeout_not_expired() {

73 - let timeout = Timeout::new(Duration::from_secs(10));

74 - assert!(!timeout.is_expired());

75 - assert!(timeout.remaining() > Duration::ZERO);

76 - }

77 -

78 - #[test]

79 - fn test_timeout_expired() {

80 - let timeout = Timeout::new(Duration::from_micros(1));

81 - thread::sleep(Duration::from_millis(10));

82 - assert!(timeout.is_expired());

83 - assert_eq!(timeout.remaining(), Duration::ZERO);

84 - }

85 -

86 - #[test]

87 - fn test_timeout_reset() {

88 - let timeout = Timeout::new(Duration::from_millis(100));

89 - thread::sleep(Duration::from_millis(60));

90 -

91 - // Should have ~40ms remaining

92 - let remaining_before = timeout.remaining();

93 - assert!(remaining_before < Duration::from_millis(50));

94 -

95 - // Reset the timeout

96 - timeout.reset();

97 -

98 - // Should now have the full initial budget (~100ms) remaining

99 - let remaining_after = timeout.remaining();

100 - assert!(remaining_after >= Duration::from_millis(90));

101 - assert!(remaining_after <= Duration::from_millis(100));

102 - }

103 -

104 - #[test]

105 - fn test_timeout_clone_shared_deadline() {

106 - let timeout1 = Timeout::new(Duration::from_secs(10));

107 - let timeout2 = timeout1.clone();

108 -

109 - thread::sleep(Duration::from_millis(100));

110 -

111 - // Reset from one clone

112 - timeout1.reset();

113 -

114 - // Both should see the reset

115 - let remaining1 = timeout1.remaining();

116 - let remaining2 = timeout2.remaining();

117 -

118 - assert!((remaining1.as_secs() as i64 - remaining2.as_secs() as i64).abs() < 1);

119 - assert!(remaining1.as_secs() >= 9); // ~10 seconds (reset to initial budget)

120 - }

121 -

122 - #[tokio::test]

123 - async fn test_tokio_timeout_with_zero_duration() {

124 - use tokio::time::timeout;

125 -

126 - // Create an expired timeout

127 - let expired_timeout = Timeout::new(Duration::from_micros(1));

128 - thread::sleep(Duration::from_millis(10));

129 - assert_eq!(expired_timeout.remaining(), Duration::ZERO);

130 -

131 - // Verify tokio::time::timeout with ZERO duration times out immediately

132 - let result = timeout(expired_timeout.remaining(), async {

133 - tokio::time::sleep(Duration::from_millis(100)).await;

134 - "should_not_complete"

135 - })

136 - .await;

137 -

138 - // Should timeout immediately

139 - assert!(result.is_err(), "Expected timeout with Duration::ZERO");

140 - }

141 -

142 - #[tokio::test]

143 - async fn test_tokio_timeout_with_remaining_time() {

144 - use tokio::time::timeout;

145 -

146 - // Create a timeout with plenty of time

147 - let valid_timeout = Timeout::new(Duration::from_secs(10));

148 - assert!(valid_timeout.remaining() > Duration::ZERO);

149 -

150 - // Verify tokio::time::timeout with remaining time completes

151 - let result = timeout(valid_timeout.remaining(), async {

152 - tokio::time::sleep(Duration::from_millis(10)).await;

153 - "completed"

154 - })

155 - .await;

156 -

157 - // Should complete successfully

158 - assert!(result.is_ok(), "Expected completion with sufficient time");

159 - assert_eq!(result.unwrap(), "completed");

160 - }

161 -}

src/crates/netdata-plugin/protocol/src/lib.rs

+2 -1

@@ -9,7 +9,8 @@ mod transport;

9 // Re-export types from netdata-plugin-types

10 pub use netdata_plugin_types::{

11 ConfigDeclaration, DynCfgCmds, DynCfgSourceType, DynCfgStatus, DynCfgType, FunctionCall,

12 - FunctionCancel, FunctionDeclaration, FunctionProgress, FunctionResult, HttpAccess,

12 + FunctionCancel, FunctionDeclaration, FunctionProgressRequest, FunctionProgressResponse,

13 + FunctionResult, HttpAccess,

14 };

16 pub use message_parser::Message;

src/crates/netdata-plugin/protocol/src/message_parser.rs

+33 -5

@@ -32,7 +32,8 @@ pub enum Message {

32 FunctionCall(Box<FunctionCall>),

33 FunctionResult(Box<FunctionResult>),

34 FunctionCancel(Box<FunctionCancel>),

35 - FunctionProgress(Box<FunctionProgress>),

35 + FunctionProgressRequest(Box<FunctionProgressRequest>),

36 + FunctionProgressResponse(Box<FunctionProgressResponse>),

37 ConfigDeclaration(Box<ConfigDeclaration>),

38 }

@@ -263,14 +264,41 @@ impl MessageParser {

264 Some(Message::FunctionCancel(function_cancel))

265 }

266

266 - /// Parse FUNCTION_PROGRESS command

267 - /// Expected format: FUNCTION_PROGRESS transaction

267 + /// Parse FUNCTION_PROGRESS command - behavior depends on parser direction

268 fn parse_function_progress(&self, args: &[u8]) -> Option<Message> {

269 + match self.direction {

270 + ParserDirection::Input => self.parse_function_progress_request(args),

271 + ParserDirection::Output => self.parse_function_progress_response(args),

272 + }

273 + }

274 +

275 + /// Parse FUNCTION_PROGRESS for input context (request from agent to plugin)

276 + /// Expected format: FUNCTION_PROGRESS transaction

277 + fn parse_function_progress_request(&self, args: &[u8]) -> Option<Message> {

278 let mut words = WordIterator::new(args);

279

280 let transaction = words.next_string()?;

272 - let function_progress = Box::new(FunctionProgress { transaction });

281

274 - Some(Message::FunctionProgress(function_progress))

282 + Some(Message::FunctionProgressRequest(Box::new(

283 + FunctionProgressRequest { transaction },

284 + )))

285 + }

286 +

287 + /// Parse FUNCTION_PROGRESS for output context (progress report from plugin to agent)

288 + /// Expected format: FUNCTION_PROGRESS transaction done all

289 + fn parse_function_progress_response(&self, args: &[u8]) -> Option<Message> {

290 + let mut words = WordIterator::new(args);

291 +

292 + let transaction = words.next_string()?;

293 + let done = words.next_usize()?;

294 + let all = words.next_usize()?;

295 +

296 + Some(Message::FunctionProgressResponse(Box::new(

297 + FunctionProgressResponse {

298 + transaction,

299 + done,

300 + all,

301 + },

302 + )))

303 }

304 }

src/crates/netdata-plugin/protocol/src/tokio_codec.rs

+13 -2

@@ -199,8 +199,19 @@ impl Encoder<Message> for MessageParser {

199 .as_bytes(),

200 );

201 }

202 - Message::FunctionProgress(_) => {

203 - unimplemented!()

202 + Message::FunctionProgressResponse(progress) => {

203 + dst.extend_from_slice(

204 + format!(

205 + "FUNCTION_PROGRESS {} {} {}\n",

206 + quote_if_needed(&progress.transaction),

207 + progress.done,

208 + progress.all,

209 + )

210 + .as_bytes(),

211 + );

212 + }

213 + Message::FunctionProgressRequest(_) => {

214 + // Inbound-only message, not encoded for output

215 }

216 }

217

src/crates/netdata-plugin/protocol/src/word_iterator.rs

@@ -50,6 +50,12 @@ impl<'a> WordIterator<'a> {

50 atoi(s)

51 }

53 + pub(crate) fn next_usize(&mut self) -> Option<usize> {

54 + let s = self.next()?;

55 +

56 + atoi(s)

57 + }

58 +

59 pub(crate) fn next_str(&mut self) -> Option<&str> {

60 let s = self.next()?;

61 std::str::from_utf8(s).ok()

src/crates/netdata-plugin/rt/Cargo.toml

-1

@@ -9,7 +9,6 @@ workspace = true

10 [dependencies]

11 # Local crates

12 -foundation = { workspace = true }

12 netdata-plugin-error = { path = "../error" }

13 netdata-plugin-protocol = { path = "../protocol" }

14 netdata-plugin-schema = { path = "../schema" }

src/crates/netdata-plugin/rt/src/lib.rs

+235 -131

@@ -23,7 +23,7 @@

23 //! use async_trait::async_trait;

24 //! use netdata_plugin_error::Result;

25 //! use netdata_plugin_protocol::FunctionDeclaration;

26 -//! use rt::{FunctionHandler, PluginRuntime};

26 +//! use rt::{FunctionCallContext, FunctionHandler, PluginRuntime};

27 //! use serde::{Deserialize, Serialize};

28 //!

29 //! #[derive(Deserialize)]

@@ -43,22 +43,16 @@

43 //! type Request = MyRequest;

44 //! type Response = MyResponse;

45 //!

46 -//! async fn on_call(&self, request: Self::Request) -> Result<Self::Response> {

46 +//! async fn on_call(

47 +//! &self,

48 +//! _ctx: FunctionCallContext,

49 +//! request: Self::Request,

50 +//! ) -> Result<Self::Response> {

51 //! Ok(MyResponse {

52 //! greeting: format!("Hello, {}!", request.name),

53 //! })

54 //! }

55 //!

52 -//! async fn on_cancellation(&self) -> Result<Self::Response> {

53 -//! Err(netdata_plugin_error::NetdataPluginError::Other {

54 -//! message: "Operation cancelled".to_string(),

55 -//! })

56 -//! }

57 -//!

58 -//! async fn on_progress(&self) {

59 -//! // Report progress if needed

60 -//! }

61 -//!

56 //! fn declaration(&self) -> FunctionDeclaration {

57 //! FunctionDeclaration::new("greet", "A greeting function")

58 //! }

@@ -97,19 +91,20 @@ use futures::future::BoxFuture;

91 use futures::stream::FuturesUnordered;

92 use netdata_plugin_error::Result;

93 use netdata_plugin_protocol::{

100 - FunctionCall, FunctionCancel, FunctionDeclaration, FunctionProgress, FunctionResult, Message,

101 - MessageReader, MessageWriter,

94 + FunctionCall, FunctionCancel, FunctionDeclaration, FunctionProgressRequest,

95 + FunctionProgressResponse, FunctionResult, Message, MessageReader, MessageWriter,

96 };

97 use serde::Serialize;

98 use serde::de::DeserializeOwned;

99 use serde_json::json;

100 use std::collections::HashMap;

101 use std::sync::Arc;

102 +use std::sync::atomic::{AtomicUsize, Ordering};

103 use std::time::Duration;

104 use tokio::io::{AsyncRead, AsyncWrite};

105 use tokio::sync::{Mutex, mpsc};

106 use tokio_util::sync::CancellationToken;

112 -use tracing::{debug, error, info, instrument, warn};

107 +use tracing::{error, info, instrument, trace, warn};

108

109 // Charts module and re-exports

110 pub mod charts;

@@ -131,39 +126,109 @@ pub use netdata_env::{LogFormat, LogLevel, LogMethod, NetdataEnv, SyslogFacility

126 mod tracing_setup;

127 pub use tracing_setup::init_tracing;

128

134 -// Re-export foundational utilities

135 -pub use foundation::Timeout;

129 +/// Atomic progress state shared between handlers and the runtime ticker.

130 +///

131 +/// Handlers write counters from any context (async, `spawn_blocking`, rayon),

132 +/// and the runtime sends progress to the agent once per second.

133 +///

134 +/// # Example

135 +///

136 +/// ```ignore

137 +/// // Set total work items before handing the counter to workers.

138 +/// ctx.progress.set_total(files.len());

139 +///

140 +/// // Give the done counter to a rayon/blocking worker.

141 +/// let counter = ctx.progress.done_counter();

142 +/// rayon::spawn(move || {

143 +/// // ... process item ...

144 +/// counter.fetch_add(1, Ordering::Relaxed);

145 +/// });

146 +///

147 +/// // Or update both at once from async code.

148 +/// ctx.progress.update(done, total);

149 +/// ```

150 +#[derive(Clone)]

151 +pub struct ProgressState {

152 + done: Arc<AtomicUsize>,

153 + total: Arc<AtomicUsize>,

154 +}

155 +

156 +impl ProgressState {

157 + fn new() -> Self {

158 + Self {

159 + done: Arc::new(AtomicUsize::new(0)),

160 + total: Arc::new(AtomicUsize::new(0)),

161 + }

162 + }

163 +

164 + /// Update both done and total. Safe from any context.

165 + pub fn update(&self, done: usize, total: usize) {

166 + self.done.store(done, Ordering::Relaxed);

167 + self.total.store(total, Ordering::Relaxed);

168 + }

169 +

170 + /// Set the total work items (e.g. before handing `done_counter` to workers).

171 + pub fn set_total(&self, total: usize) {

172 + self.total.store(total, Ordering::Relaxed);

173 + }

174 +

175 + /// Get a clone of the done counter for sharing with worker threads.

176 + /// Workers call `counter.fetch_add(1, Ordering::Relaxed)` directly.

177 + pub fn done_counter(&self) -> Arc<AtomicUsize> {

178 + self.done.clone()

179 + }

180 +

181 + fn load(&self) -> (usize, usize) {

182 + (

183 + self.done.load(Ordering::Relaxed),

184 + self.total.load(Ordering::Relaxed),

185 + )

186 + }

187 +}

188 +

189 +/// Context provided to function handlers during execution.

190 +///

191 +/// Contains the transaction identifier, atomic progress state, and a

192 +/// cancellation token that signals when the function should stop.

193 +pub struct FunctionCallContext {

194 + /// Unique identifier for this function call.

195 + transaction: String,

196 + /// Atomic progress state. The runtime reads these counters once per

197 + /// second and sends progress to the agent automatically.

198 + pub progress: ProgressState,

199 + /// Token that signals when the function should stop.

200 + /// Check `is_cancelled()` in sync code, or `await cancelled()` in async code.

201 + pub cancellation: CancellationToken,

202 +}

203

137 -/// Internal control signals sent to running functions.

138 -enum RuntimeSignal {

139 - /// Signal to request progress update from a running function.

140 - Progress,

204 +impl FunctionCallContext {

205 + /// Returns the transaction identifier for this function call.

206 + pub fn transaction(&self) -> &str {

207 + &self.transaction

208 + }

209 }

210

211 /// Represents an active function call transaction.

212 ///

213 /// Each transaction tracks a single function invocation, including its

146 -/// unique identifier, control channel for signals, and cancellation token.

214 +/// unique identifier and cancellation token.

215 struct Transaction {

216 /// Unique identifier for this transaction.

217 id: String,

150 - /// Channel for sending control signals to the running function.

151 - control_tx: mpsc::Sender<RuntimeSignal>,

218 /// Token for cancelling this specific function execution.

219 cancellation_token: CancellationToken,

220 }

221

156 -/// Execution context provided to function handlers.

222 +/// Execution context provided to the handler adapter layer.

223 ///

158 -/// Contains all the information and control mechanisms needed for

159 -/// a function to execute, handle cancellation, and report progress.

224 +/// Contains all the information needed for a function to execute.

225 struct FunctionContext {

226 /// The original function call request from Netdata.

227 function_call: Box<FunctionCall>,

228 /// Token for detecting cancellation requests.

229 cancellation_token: CancellationToken,

165 - /// Receiver for runtime control signals (e.g., progress requests).

166 - signal_rx: Mutex<mpsc::Receiver<RuntimeSignal>>,

230 + /// Sender for outbound messages (e.g., progress reports back to the agent).

231 + outbound_tx: mpsc::UnboundedSender<Message>,

232 }

233

234 /// Type alias for a future that produces a function result.

@@ -205,22 +270,16 @@ type FunctionFuture = BoxFuture<'static, (String, FunctionResult)>;

270 /// type Request = AddRequest;

271 /// type Response = AddResponse;

272 ///

208 -/// async fn on_call(&self, request: Self::Request) -> Result<Self::Response> {

273 +/// async fn on_call(

274 +/// &self,

275 +/// _ctx: FunctionCallContext,

276 +/// request: Self::Request,

277 +/// ) -> Result<Self::Response> {

278 /// Ok(AddResponse {

279 /// sum: request.a + request.b,

280 /// })

281 /// }

282 ///

214 -/// async fn on_cancellation(&self) -> Result<Self::Response> {

215 -/// Err(netdata_plugin_error::NetdataPluginError::Other {

216 -/// message: "Addition cancelled".to_string(),

217 -/// })

218 -/// }

219 -///

220 -/// async fn on_progress(&self) {

221 -/// // Not needed for quick operations

222 -/// }

223 -///

283 /// fn declaration(&self) -> FunctionDeclaration {

284 /// FunctionDeclaration::new("add", "Adds two numbers")

285 /// }

@@ -243,11 +302,13 @@ pub trait FunctionHandler: Send + Sync + 'static {

302 /// Main function logic executed when the function is called.

303 ///

304 /// This method contains the primary computation or operation that the

246 - /// function performs. It receives the deserialized request and should

247 - /// return either a successful response or an error.

305 + /// function performs. It receives the deserialized request, a context

306 + /// for progress reporting and cancellation, and should return either

307 + /// a successful response or an error.

308 ///

309 /// # Arguments

310 ///

311 + /// * `ctx` - Context with transaction ID, progress sender, and cancellation token

312 /// * `request` - The deserialized request payload

313 ///

314 /// # Returns

@@ -256,32 +317,14 @@ pub trait FunctionHandler: Send + Sync + 'static {

317 ///

318 /// # Cancellation

319 ///

259 - /// This method may be interrupted if a cancellation is requested.

260 - /// When cancelled, the runtime will call [`on_cancellation`](Self::on_cancellation) instead.

261 - async fn on_call(&self, transaction: String, request: Self::Request) -> Result<Self::Response>;

262 -

263 - /// Handle cancellation requests while the function is running.

264 - ///

265 - /// Called when Netdata requests cancellation of a running function.

266 - /// This method should quickly return an appropriate error or partial result.

267 - ///

268 - /// # Returns

269 - ///

270 - /// Typically returns an error indicating the operation was cancelled,

271 - /// but may return a partial result if appropriate.

272 - async fn on_cancellation(&self, transaction: String) -> Result<Self::Response>;

273 -

274 - /// Handle progress report requests while the function is running.

275 - ///

276 - /// Called when Netdata requests a progress update from a long-running function.

277 - /// This method should log or report the current progress but doesn't need

278 - /// to return a value (progress is typically reported through logging).

279 - ///

280 - /// # Note

281 - ///

282 - /// This is called asynchronously while `on_call` is still running,

283 - /// so any shared state must be properly synchronized.

284 - async fn on_progress(&self, transaction: String);

320 + /// When cancelled, the runtime cancels the token in `ctx.cancellation`

321 + /// and drops this future. Check `ctx.cancellation.is_cancelled()` in

322 + /// synchronous code paths.

323 + async fn on_call(

324 + &self,

325 + ctx: FunctionCallContext,

326 + request: Self::Request,

327 + ) -> Result<Self::Response>;

328

329 /// Provide the function's declaration metadata.

330 ///

@@ -362,33 +405,62 @@ impl<H: FunctionHandler> RawFunctionHandler for HandlerAdapter<H> {

405 },

406 };

407

365 - // Drive the handler with cancellation and progress handling

366 - let handler = self.handler.clone();

367 -

368 - let mut call_future = Box::pin(handler.on_call(transaction.clone(), payload));

369 - let mut signal_rx = ctx.signal_rx.lock().await;

408 + // Build the function call context

409 + let call_ctx = FunctionCallContext {

410 + transaction: transaction.clone(),

411 + progress: ProgressState::new(),

412 + cancellation: ctx.cancellation_token.clone(),

413 + };

414

371 - let result = loop {

372 - tokio::select! {

373 - // Poll the main computation

374 - result = &mut call_future => {

375 - break result;

376 - }

377 - // Handle progress requests

378 - Some(msg) = signal_rx.recv() => {

379 - match msg {

380 - RuntimeSignal::Progress => {

381 - handler.on_progress(transaction.clone()).await;

382 - }

415 + // Spawn a background ticker that reads the atomic progress counters

416 + // once per second and sends FunctionProgressResponse to the agent.

417 + let progress = call_ctx.progress.clone();

418 + let ticker_tx = ctx.outbound_tx.clone();

419 + let ticker_transaction = transaction.clone();

420 +

421 + let ticker = tokio::spawn(async move {

422 + let mut interval = tokio::time::interval(Duration::from_secs(1));

423 + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

424 + loop {

425 + interval.tick().await;

426 + let (done, total) = progress.load();

427 + if total > 0 {

428 + let msg =

429 + Message::FunctionProgressResponse(Box::new(FunctionProgressResponse {

430 + transaction: ticker_transaction.clone(),

431 + done,

432 + all: total,

433 + }));

434 + tracing::trace!(

435 + "[{}] progress {}/{}",

436 + ticker_transaction.clone(),

437 + done,

438 + total

439 + );

440 + if ticker_tx.send(msg).is_err() {

441 + tracing::error!(

442 + "[{}] outbound channel closed, stopping progress ticker",

443 + ticker_transaction

444 + );

445 + break;

446 }

447 }

385 - // Handle cancellation

386 - _ = ctx.cancellation_token.cancelled() => {

387 - break handler.on_cancellation(transaction.clone()).await;

388 - }

448 + }

449 + });

450 +

451 + let handler = self.handler.clone();

452 +

453 + let result = tokio::select! {

454 + result = handler.on_call(call_ctx, payload) => result,

455 + _ = ctx.cancellation_token.cancelled() => {

456 + Err(netdata_plugin_error::NetdataPluginError::Other {

457 + message: "Function cancelled".to_string(),

458 + })

459 }

460 };

461

462 + ticker.abort();

463 +

464 let current_timestamp = std::time::SystemTime::now()

465 .duration_since(std::time::UNIX_EPOCH)

466 .expect("Time went backwards")

@@ -421,7 +493,11 @@ impl<H: FunctionHandler> RawFunctionHandler for HandlerAdapter<H> {

493 }

494 }

495 Err(e) => {

424 - error!("function handler error: {}", e);

496 + if ctx.cancellation_token.is_cancelled() {

497 + info!("function handler cancelled: {}", e);

498 + } else {

499 + error!("function handler error: {}", e);

500 + }

501 let error_json = json!({

502 "error": format!("{}", e),

503 "status": 500

@@ -501,6 +577,11 @@ where

577 /// Token for initiating graceful shutdown.

578 shutdown_token: CancellationToken,

579

580 + /// Sender for outbound messages (progress reports, function results).

581 + outbound_tx: mpsc::UnboundedSender<Message>,

582 + /// Receiver for outbound messages — consumed by the writer task.

583 + outbound_rx: Option<mpsc::UnboundedReceiver<Message>>,

584 +

585 /// Optional chart registry for managing metrics emission.

586 chart_registry: Option<ChartRegistry<W>>,

587 /// Handle to chart registry background task.

@@ -566,6 +647,8 @@ where

647 /// }

648 /// ```

649 pub fn with_streams(name: &str, reader: R, writer: W) -> Self {

650 + let (outbound_tx, outbound_rx) = mpsc::unbounded_channel();

651 +

652 Self {

653 plugin_name: String::from(name),

654 reader: MessageReader::new(reader),

@@ -576,6 +659,8 @@ where

659 futures: FuturesUnordered::new(),

660

661 shutdown_token: CancellationToken::default(),

662 + outbound_tx,

663 + outbound_rx: Some(outbound_rx),

664 chart_registry: None,

665 chart_registry_handle: None,

666 }

@@ -730,9 +815,50 @@ where

815 }

816

817 self.declare_functions().await?;

818 +

819 + // Spawn a dedicated writer task so stdout I/O never blocks the

820 + // main select loop (which must keep reading stdin).

821 + let writer = Arc::clone(&self.writer);

822 + let mut outbound_rx = self

823 + .outbound_rx

824 + .take()

825 + .expect("outbound_rx consumed only once");

826 +

827 + let writer_task = tokio::spawn(async move {

828 + let mut keepalive =

829 + tokio::time::interval(tokio::time::Duration::from_secs(60));

830 +

831 + loop {

832 + tokio::select! {

833 + msg = outbound_rx.recv() => {

834 + match msg {

835 + Some(msg) => {

836 + if let Err(e) = writer.lock().await.send(msg).await {

837 + error!("outbound writer error: {}", e);

838 + break;

839 + }

840 + }

841 + None => break,

842 + }

843 + }

844 + _ = keepalive.tick() => {

845 + if let Err(e) = writer.lock().await.write_raw(b"PLUGIN_KEEPALIVE\n").await {

846 + error!("keepalive write error: {}", e);

847 + break;

848 + }

849 + }

850 + }

851 + }

852 + });

853 +

854 self.process_messages().await?;

855 self.shutdown().await?;

856

857 + // All outbound senders (including those in handler contexts) are now

858 + // dropped, so the writer task will drain and exit.

859 + drop(self);

860 + let _ = writer_task.await;

861 +

862 Ok(())

863 }

864

@@ -805,20 +931,18 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

931

932 loop {

933 tokio::select! {

808 - // Make the shutdown signal higher priority by putting it first

934 _ = self.shutdown_token.cancelled() => {

810 - info!("shutdown requested... Stop processing messages from stdin");

811 - // Reader will be dropped here, closing stdin

935 + info!("shutdown requested, stop processing messages from stdin");

936 break;

937 }

938 + Some((transaction, result)) = self.futures.next() => {

939 + self.handle_completed(transaction, result).await?;

940 + }

941 message = self.reader.next() => {

942 if self.handle_message(message).await? {

943 break;

944 }

945 }

819 - Some((transaction, result)) = self.futures.next() => {

820 - self.handle_completed(transaction, result).await?;

821 - }

946 }

947 }

948

@@ -838,11 +962,11 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

962 Some(Ok(Message::FunctionCancel(function_cancel))) => {

963 self.handle_function_cancel(function_cancel.as_ref());

964 }

841 - Some(Ok(Message::FunctionProgress(function_progress))) => {

842 - self.handle_function_progress(&function_progress).await;

965 + Some(Ok(Message::FunctionProgressRequest(req))) => {

966 + trace!(transaction = %req.transaction, "ignoring inbound progress request");

967 }

968 Some(Ok(msg)) => {

845 - debug!("received message: {:?}", msg);

969 + trace!("received message: {:?}", msg);

970 }

971 Some(Err(e)) => {

972 error!("error parsing message: {:?}", e);

@@ -850,9 +974,10 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

974 None => {

975 info!("input stream ended");

976 self.shutdown_token.cancel();

853 - return Ok(true); // Signal to break the loop

977 + return Ok(true);

978 }

979 }

980 +

981 Ok(false)

982 }

983

@@ -907,13 +1032,12 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

1032 };

1033

1034 // Create a new function context

910 - let (control_tx, control_rx) = mpsc::channel(4);

1035 let cancellation_token = CancellationToken::new();

1036

1037 let function_context = Arc::new(FunctionContext {

1038 function_call,

1039 cancellation_token: cancellation_token.clone(),

916 - signal_rx: Mutex::new(control_rx),

1040 + outbound_tx: self.outbound_tx.clone(),

1041 });

1042

1043 // Create new transaction

@@ -921,7 +1045,6 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

1045 let transaction = Arc::new(Transaction {

1046 id,

1047 cancellation_token,

924 - control_tx,

1048 });

1049 self.transaction_registry

1050 .insert(transaction.id.clone(), transaction.clone());

@@ -950,42 +1073,23 @@ impl<R: AsyncRead + Unpin + Send, W: AsyncWrite + Unpin + Send> PluginRuntime<R,

1073 transaction.cancellation_token.cancel();

1074 }

1075

953 - /// Handle a progress report request.

954 - ///

955 - /// Sends a progress signal to the corresponding running function.

956 - async fn handle_function_progress(&mut self, function_progress: &FunctionProgress) {

957 - let Some(transaction) = self

958 - .transaction_registry

959 - .get(&function_progress.transaction)

960 - else {

961 - warn!(

962 - "can not get progress of non-existing transaction {}",

963 - function_progress.transaction

964 - );

965 - return;

966 - };

967 -

968 - info!(

969 - "requesting progress of transaction {}",

970 - function_progress.transaction

971 - );

972 - let _ = transaction.control_tx.send(RuntimeSignal::Progress).await;

973 - }

974 -

1076 /// Handle a completed function execution.

1077 ///

977 - /// Removes the transaction from the registry and sends the result back to Netdata.

1078 + /// Removes the transaction from the registry and sends the result

1079 + /// through the outbound channel (written to stdout by the writer task).

1080 async fn handle_completed(

1081 &mut self,

1082 transaction: String,

1083 result: FunctionResult,

1084 ) -> Result<()> {

1085 self.transaction_registry.remove(&transaction);

984 - self.writer

985 - .lock()

986 - .await

987 - .send(Message::FunctionResult(Box::new(result)))

988 - .await?;

1086 + let msg = Message::FunctionResult(Box::new(result));

1087 + if self.outbound_tx.send(msg).is_err() {

1088 + error!(

1089 + "outbound channel closed, cannot send result for transaction {}",

1090 + transaction

1091 + );

1092 + }

1093 Ok(())

1094 }

1095

src/crates/netdata-plugin/rt/src/tracing_setup.rs

+5 -3

@@ -47,7 +47,7 @@ fn log_level_to_filter(level: &LogLevel) -> &'static str {

47 LogLevel::Error => "error",

48 LogLevel::Warning => "warn",

49 LogLevel::Notice | LogLevel::Info => "info",

50 - LogLevel::Debug => "debug",

50 + LogLevel::Debug => "trace",

51 }

52 }

@@ -68,8 +68,10 @@ pub fn init_tracing() {

68 .map(log_level_to_filter)

69 .unwrap_or("info");

71 - // Create environment filter

72 - let env_filter = EnvFilter::new(filter_str);

71 + // Create environment filter: configured level as default, but limit noisy

72 + // third-party crates to warn.

73 + let filter = format!("{filter_str},foyer=warn,notify=warn");

74 + let env_filter = EnvFilter::new(&filter);

76 // Build the registry with base layers

77 let registry = tracing_subscriber::registry().with(env_filter);

src/crates/netdata-plugin/types/src/functions.rs

+14 -3

@@ -77,9 +77,20 @@ pub struct FunctionCancel {

77 pub transaction: String,

78 }

80 -/// A message for reporting function call progress

80 +/// A request from the agent for a progress report on a running function call

81 #[derive(Debug, Clone)]

82 -pub struct FunctionProgress {

83 - /// Transaction ID of the function call that should report the progress

82 +pub struct FunctionProgressRequest {

83 + /// Transaction ID of the function call to report progress for

84 pub transaction: String,

85 }

86 +

87 +/// A progress report sent from the plugin to the agent

88 +#[derive(Debug, Clone)]

89 +pub struct FunctionProgressResponse {

90 + /// Transaction ID of the function call reporting progress

91 + pub transaction: String,

92 + /// Number of units completed

93 + pub done: usize,

94 + /// Total number of units to complete

95 + pub all: usize,

96 +}

src/crates/netdata-plugin/types/src/lib.rs

+2 -1

@@ -18,6 +18,7 @@ pub use dyncfg_status::DynCfgStatus;

18 pub use dyncfg_type::DynCfgType;

20 pub use functions::{

21 - FunctionCall, FunctionCancel, FunctionDeclaration, FunctionProgress, FunctionResult,

21 + FunctionCall, FunctionCancel, FunctionDeclaration, FunctionProgressRequest,

22 + FunctionProgressResponse, FunctionResult,

23 };

24 pub use http_access::HttpAccess;

src/plugins.d/pluginsd_functions.c

@@ -130,6 +130,12 @@ void pluginsd_inflight_functions_garbage_collect(PARSER *parser, usec_t now_ut)

130 "Timeout waiting for a response.",

131 HTTP_RESP_GATEWAY_TIMEOUT);

132

133 + // Notify the plugin that the transaction has been cancelled due to timeout,

134 + // so it can stop any in-progress work for this transaction.

135 + char buffer[2048];

136 + snprintfz(buffer, sizeof(buffer), PLUGINSD_CALL_FUNCTION_CANCEL " %s\n", pf_dfe.name);

137 + send_to_plugin(buffer, pf->parser, STREAM_TRAFFIC_TYPE_FUNCTIONS);

138 +

139 dictionary_del(parser->inflight.functions, pf_dfe.name);

140 }

141