Optionally limit the number of retrieved events

disk/src/binned.rs

@@ -351,7 +351,7 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
                     self.query.disk_stats_every().clone(),
                     self.query.report_error(),
                 )?;
-                let f = collect_plain_events_json(s, self.timeout, t_bin_count, self.query.do_log());
+                let f = collect_plain_events_json(s, self.timeout, t_bin_count, u64::MAX, self.query.do_log());
                 let s = futures_util::stream::once(f).map(|item| match item {
                     Ok(item) => Ok(Bytes::from(serde_json::to_vec(&item)?)),
                     Err(e) => Err(e.into()),
@@ -375,7 +375,7 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
                     x_bin_count,
                     self.query.agg_kind().do_time_weighted(),
                 );
-                let f = collect_plain_events_json(s, self.timeout, t_bin_count, self.query.do_log());
+                let f = collect_plain_events_json(s, self.timeout, t_bin_count, u64::MAX, self.query.do_log());
                 let s = futures_util::stream::once(f).map(|item| match item {
                     Ok(item) => Ok(Bytes::from(serde_json::to_vec(&item)?)),
                     Err(e) => Err(e.into()),

disk/src/channelexec.rs

@@ -208,7 +208,9 @@ where
         range: range.clone(),
         expand: agg_kind.need_expand(),
     };
-    let conf = httpclient::get_channel_config(&q, node_config).await?;
+    let conf = httpclient::get_channel_config(&q, node_config)
+        .await
+        .map_err(|e| e.add_public_msg(format!("Can not find channel config for {}", q.channel.name())))?;
     let ret = channel_exec_config(
         f,
         conf.scalar_type.clone(),
@@ -293,6 +295,7 @@ pub struct PlainEventsJson {
     disk_io_buffer_size: usize,
     timeout: Duration,
     node_config: NodeConfigCached,
+    events_max: u64,
     do_log: bool,
 }
 
@@ -303,6 +306,7 @@ impl PlainEventsJson {
         disk_io_buffer_size: usize,
         timeout: Duration,
         node_config: NodeConfigCached,
+        events_max: u64,
         do_log: bool,
     ) -> Self {
         Self {
@@ -312,6 +316,7 @@ impl PlainEventsJson {
             disk_io_buffer_size,
             timeout,
             node_config,
+            events_max,
             do_log,
         }
     }
@@ -330,6 +335,7 @@ pub async fn collect_plain_events_json<T, S>(
     stream: S,
     timeout: Duration,
     bin_count_exp: u32,
+    events_max: u64,
     do_log: bool,
 ) -> Result<JsonValue, Error>
 where
@@ -394,6 +400,9 @@ where
                             RangeCompletableItem::Data(item) => {
                                 collector.ingest(&item);
                                 i1 += 1;
+                                if i1 >= events_max {
+                                    break;
+                                }
                             }
                         },
                     },
@@ -448,7 +457,7 @@ impl ChannelExecFunction for PlainEventsJson {
             do_decompress: true,
         };
         let s = MergedFromRemotes::<ENP>::new(evq, perf_opts, self.node_config.node_config.cluster);
-        let f = collect_plain_events_json(s, self.timeout, 0, self.do_log);
+        let f = collect_plain_events_json(s, self.timeout, 0, self.events_max, self.do_log);
         let f = FutureExt::map(f, |item| match item {
             Ok(item) => {
                 // TODO add channel entry info here?
@@ -466,14 +475,3 @@ impl ChannelExecFunction for PlainEventsJson {
         Box::pin(futures_util::stream::empty())
     }
 }
-
-// TODO remove when done.
-pub fn dummy_impl() {
-    let channel: Channel = err::todoval();
-    let range: NanoRange = err::todoval();
-    let agg_kind: AggKind = err::todoval();
-    let node_config: NodeConfigCached = err::todoval();
-    let timeout: Duration = err::todoval();
-    let f = PlainEventsJson::new(channel.clone(), range.clone(), 0, timeout, node_config.clone(), false);
-    let _ = channel_exec(f, &channel, &range, agg_kind, &node_config);
-}

disk/src/events.rs

@@ -109,17 +109,25 @@ pub struct PlainEventsJsonQuery {
     disk_io_buffer_size: usize,
     report_error: bool,
     timeout: Duration,
+    events_max: Option<u64>,
     do_log: bool,
 }
 
 impl PlainEventsJsonQuery {
-    pub fn new(channel: Channel, range: NanoRange, disk_io_buffer_size: usize, do_log: bool) -> Self {
+    pub fn new(
+        channel: Channel,
+        range: NanoRange,
+        disk_io_buffer_size: usize,
+        events_max: Option<u64>,
+        do_log: bool,
+    ) -> Self {
         Self {
             channel,
             range,
             disk_io_buffer_size,
             report_error: false,
             timeout: Duration::from_millis(10000),
+            events_max,
             do_log,
         }
     }
@@ -150,6 +158,9 @@ impl PlainEventsJsonQuery {
                 .parse::<u64>()
                 .map(|k| Duration::from_millis(k))
                 .map_err(|e| Error::with_public_msg(format!("can not parse timeout {:?}", e)))?,
+            events_max: pairs
+                .get("eventsMax")
+                .map_or(Ok(None), |k| k.parse().map(|k| Some(k)))?,
             do_log: pairs
                 .get("doLog")
                 .map_or("false", |k| k)
@@ -185,6 +196,10 @@ impl PlainEventsJsonQuery {
         self.timeout
     }
 
+    pub fn events_max(&self) -> Option<u64> {
+        self.events_max
+    }
+
     pub fn do_log(&self) -> bool {
         self.do_log
     }
@@ -208,6 +223,9 @@ impl PlainEventsJsonQuery {
         );
         g.append_pair("diskIoBufferSize", &format!("{}", self.disk_io_buffer_size));
         g.append_pair("timeout", &format!("{}", self.timeout.as_millis()));
+        if let Some(x) = self.events_max.as_ref() {
+            g.append_pair("eventsMax", &format!("{}", x));
+        }
         g.append_pair("doLog", &format!("{}", self.do_log));
     }
 }