revert counts request changes

Author: parmesant

src/prism/logstream/mod.rs

@@ -20,35 +20,28 @@ use std::sync::Arc;
 
 use actix_web::http::header::ContentType;
 use arrow_schema::Schema;
-use chrono::{TimeDelta, Utc};
+use chrono::Utc;
 use http::StatusCode;
-use itertools::Itertools;
 use serde::{Deserialize, Serialize};
-use serde_json::{Value, json};
 use tracing::warn;
 
 use crate::{
     LOCK_EXPECT,
-    alerts::alert_structs::{ConditionConfig, Conditions},
-    event::DEFAULT_TIMESTAMP_KEY,
     handlers::http::{
         cluster::{
             fetch_stats_from_ingestors,
             utils::{IngestionStats, QueriedStats, StorageStats, merge_queried_stats},
         },
         logstream::error::StreamError,
-        query::{Query, QueryError, get_records_and_fields, update_schema_when_distributed},
+        query::{QueryError, update_schema_when_distributed},
     },
     hottier::{HotTierError, HotTierManager, StreamHotTier},
     parseable::{PARSEABLE, StreamNotFound},
-    query::{CountConditions, CountsRequest, CountsResponse, error::ExecuteError},
+    query::{CountsRequest, CountsResponse, error::ExecuteError},
     rbac::{Users, map::SessionKey, role::Action},
     stats,
     storage::{StreamInfo, StreamType, retention::Retention},
-    utils::{
-        arrow::record_batches_to_json,
-        time::{TimeParseError, truncate_to_minute},
-    },
+    utils::time::TimeParseError,
     validator::error::HotTierValidationError,
 };
 
@@ -299,7 +292,7 @@ impl PrismDatasetRequest {
 
         // Process stream data
         match get_prism_logstream_info(&stream).await {
-            Ok(info) => Ok(Some(self.build_dataset_response(stream, info, &key).await?)),
+            Ok(info) => Ok(Some(self.build_dataset_response(stream, info).await?)),
             Err(err) => Err(err),
         }
     }
@@ -319,13 +312,12 @@ impl PrismDatasetRequest {
         &self,
         stream: String,
         info: PrismLogstreamInfo,
-        key: &SessionKey,
     ) -> Result<PrismDatasetResponse, PrismLogstreamError> {
         // Get hot tier info
         let hottier = self.get_hot_tier_info(&stream).await?;
 
         // Get counts
-        let counts = self.get_counts(&stream, key).await?;
+        let counts = self.get_counts(&stream).await?;
 
         Ok(PrismDatasetResponse {
             stream,
@@ -354,84 +346,20 @@ impl PrismDatasetRequest {
         }
     }
 
-    async fn get_counts(
-        &self,
-        stream: &str,
-        key: &SessionKey,
-    ) -> Result<CountsResponse, PrismLogstreamError> {
-        let end = truncate_to_minute(Utc::now());
-        let start = end - TimeDelta::hours(1);
-
-        let conditions = if PARSEABLE.get_stream(stream)?.get_time_partition().is_some() {
-            Some(CountConditions {
-                conditions: Some(Conditions {
-                    operator: Some(crate::alerts::LogicalOperator::And),
-                    condition_config: vec![
-                        ConditionConfig {
-                            column: DEFAULT_TIMESTAMP_KEY.into(),
-                            operator: crate::alerts::WhereConfigOperator::GreaterThanOrEqual,
-                            value: Some(start.to_rfc3339()),
-                        },
-                        ConditionConfig {
-                            column: DEFAULT_TIMESTAMP_KEY.into(),
-                            operator: crate::alerts::WhereConfigOperator::LessThan,
-                            value: Some(end.to_rfc3339()),
-                        },
-                    ],
-                }),
-                group_by: None,
-            })
-        } else {
-            None
-        };
-
+    async fn get_counts(&self, stream: &str) -> Result<CountsResponse, PrismLogstreamError> {
         let count_request = CountsRequest {
             stream: stream.to_owned(),
-            start_time: start.to_rfc3339(),
-            end_time: end.to_rfc3339(),
+            start_time: "1h".to_owned(),
+            end_time: "now".to_owned(),
             num_bins: 10,
-            conditions,
+            conditions: None,
         };
 
-        if count_request.conditions.is_some() {
-            // forward request to querier
-            let query = count_request
-                .get_df_sql(DEFAULT_TIMESTAMP_KEY.into())
-                .await?;
-
-            let query_request = Query {
-                query,
-                start_time: start.to_rfc3339(),
-                end_time: end.to_rfc3339(),
-                send_null: true,
-                fields: true,
-                streaming: false,
-                filter_tags: None,
-            };
-
-            let (records, _) = get_records_and_fields(&query_request, key).await?;
-            if let Some(records) = records {
-                let json_records = record_batches_to_json(&records)?;
-                let records = json_records.into_iter().map(Value::Object).collect_vec();
-
-                let res = json!({
-                    "fields": vec!["start_time", "end_time", "count"],
-                    "records": records,
-                });
-
-                Ok(serde_json::from_value(res)?)
-            } else {
-                Err(PrismLogstreamError::Anyhow(anyhow::Error::msg(
-                    "No data returned for counts SQL",
-                )))
-            }
-        } else {
-            let records = count_request.get_bin_density().await?;
-            Ok(CountsResponse {
-                fields: vec!["start_time".into(), "end_time".into(), "count".into()],
-                records,
-            })
-        }
+        let records = count_request.get_bin_density().await?;
+        Ok(CountsResponse {
+            fields: vec!["start_time".into(), "end_time".into(), "count".into()],
+            records,
+        })
     }
 }