@@ -39,7 +39,7 @@ use crate::{
39
39
utils:: time:: TimeRange ,
40
40
} ;
41
41
42
- use super :: { AlertConfig , AlertError } ;
42
+ use super :: { AlertConfig , AlertError , ThresholdConfig } ;
43
43
44
44
async fn get_tables_from_query ( query : & str ) -> Result < TableScanVisitor , AlertError > {
45
45
let session_state = QUERY_SESSION . state ( ) ;
@@ -120,16 +120,37 @@ pub async fn evaluate_alert(alert: &AlertConfig) -> Result<(), AlertError> {
120
120
. await
121
121
. map_err ( |err| AlertError :: CustomError ( err. to_string ( ) ) ) ?;
122
122
123
- // let df = DataFrame::new(session_state, raw_logical_plan);
123
+ let ( group_expr, aggr_expr, filter_expr) = get_exprs ( & alert. thresholds ) ;
124
+ let df = df. aggregate ( group_expr, aggr_expr) ?;
125
+
126
+ let nrows = df. clone ( ) . filter ( filter_expr) ?. count ( ) . await ?;
127
+ trace ! ( "dataframe-\n {:?}" , df. collect( ) . await ) ;
128
+
129
+ if nrows > 0 {
130
+ trace ! ( "ALERT!!!!!!" ) ;
131
+
132
+ // update state
133
+ ALERTS
134
+ . update_state ( & alert. id . to_string ( ) , AlertState :: Triggered , true )
135
+ . await ?;
136
+ } else {
137
+ ALERTS
138
+ . update_state ( & alert. id . to_string ( ) , AlertState :: Resolved , false )
139
+ . await ?;
140
+ }
141
+
142
+ Ok ( ( ) )
143
+ }
124
144
145
+ fn get_exprs ( thresholds : & Vec < ThresholdConfig > ) -> ( Vec < Expr > , Vec < Expr > , Expr ) {
125
146
// for now group by is empty, we can include this later
126
- let group_expr = vec ! [ ] ;
147
+ let group_expr: Vec < Expr > = vec ! [ ] ;
127
148
128
149
// agg expression
129
- let mut aggr_expr = vec ! [ ] ;
150
+ let mut aggr_expr: Vec < Expr > = vec ! [ ] ;
130
151
131
152
let mut expr = Expr :: Literal ( datafusion:: scalar:: ScalarValue :: Boolean ( Some ( true ) ) ) ;
132
- for threshold in & alert . thresholds {
153
+ for threshold in thresholds {
133
154
let res = match threshold. operator {
134
155
crate :: handlers:: http:: alerts:: AlertOperator :: GreaterThan => {
135
156
col ( & threshold. column ) . gt ( lit ( threshold. value ) )
@@ -177,23 +198,5 @@ pub async fn evaluate_alert(alert: &AlertConfig) -> Result<(), AlertError> {
177
198
expr = expr. and ( res) ;
178
199
}
179
200
180
- let df = df. aggregate ( group_expr, aggr_expr) ?;
181
-
182
- let nrows = df. clone ( ) . filter ( expr) ?. count ( ) . await ?;
183
- trace ! ( "dataframe-\n {:?}" , df. collect( ) . await ) ;
184
-
185
- if nrows > 0 {
186
- trace ! ( "ALERT!!!!!!" ) ;
187
-
188
- // update state
189
- ALERTS
190
- . update_state ( & alert. id . to_string ( ) , AlertState :: Triggered , true )
191
- . await ?;
192
- } else {
193
- ALERTS
194
- . update_state ( & alert. id . to_string ( ) , AlertState :: Resolved , false )
195
- . await ?;
196
- }
197
-
198
- Ok ( ( ) )
201
+ ( group_expr, aggr_expr, expr)
199
202
}
0 commit comments