@@ -577,10 +577,12 @@ impl AlertConfig {
577
577
store : & dyn crate :: storage:: ObjectStorage ,
578
578
) -> Result < AlertConfig , AlertError > {
579
579
let basic_fields = Self :: parse_basic_fields ( alert_json) ?;
580
- let query = Self :: build_query_from_v1 ( alert_json) . await ?;
581
- let threshold_config = Self :: extract_threshold_config ( alert_json) ?;
582
- let eval_config = Self :: extract_eval_config ( alert_json) ?;
583
- let targets = Self :: extract_targets ( alert_json) ?;
580
+ let alert_info = format ! ( "Alert '{}' (ID: {})" , basic_fields. title, basic_fields. id) ;
581
+
582
+ let query = Self :: build_query_from_v1 ( alert_json, & alert_info) . await ?;
583
+ let threshold_config = Self :: extract_threshold_config ( alert_json, & alert_info) ?;
584
+ let eval_config = Self :: extract_eval_config ( alert_json, & alert_info) ?;
585
+ let targets = Self :: extract_targets ( alert_json, & alert_info) ?;
584
586
let state = Self :: extract_state ( alert_json) ;
585
587
586
588
// Create the migrated v2 alert
@@ -613,12 +615,14 @@ impl AlertConfig {
613
615
614
616
let title = alert_json[ "title" ]
615
617
. as_str ( )
616
- . ok_or_else ( || AlertError :: CustomError ( "Missing title in v1 alert" . to_string ( ) ) ) ?
618
+ . ok_or_else ( || {
619
+ AlertError :: CustomError ( format ! ( "Missing title in v1 alert (ID: {id})" ) )
620
+ } ) ?
617
621
. to_string ( ) ;
618
622
619
- let severity_str = alert_json[ "severity" ]
620
- . as_str ( )
621
- . ok_or_else ( || AlertError :: CustomError ( "Missing severity in v1 alert" . to_string ( ) ) ) ?;
623
+ let severity_str = alert_json[ "severity" ] . as_str ( ) . ok_or_else ( || {
624
+ AlertError :: CustomError ( format ! ( "Missing severity in v1 alert '{title}' (ID: {id})" ) )
625
+ } ) ?;
622
626
623
627
let severity = match severity_str. to_lowercase ( ) . as_str ( ) {
624
628
"critical" => Severity :: Critical ,
@@ -636,30 +640,38 @@ impl AlertConfig {
636
640
}
637
641
638
642
/// Build SQL query from v1 alert structure
639
- async fn build_query_from_v1 ( alert_json : & JsonValue ) -> Result < String , AlertError > {
640
- let stream = alert_json[ "stream" ]
641
- . as_str ( )
642
- . ok_or_else ( || AlertError :: CustomError ( "Missing stream in v1 alert" . to_string ( ) ) ) ?;
643
+ async fn build_query_from_v1 (
644
+ alert_json : & JsonValue ,
645
+ alert_info : & str ,
646
+ ) -> Result < String , AlertError > {
647
+ let stream = alert_json[ "stream" ] . as_str ( ) . ok_or_else ( || {
648
+ AlertError :: CustomError ( format ! ( "Missing stream in v1 alert for {alert_info}" ) )
649
+ } ) ?;
643
650
644
651
let aggregates = & alert_json[ "aggregates" ] ;
645
652
let aggregate_config = & aggregates[ "aggregateConfig" ] [ 0 ] ;
646
653
647
- let aggregate_function = Self :: parse_aggregate_function ( aggregate_config) ?;
648
- let base_query = Self :: build_base_query ( & aggregate_function, aggregate_config, stream) ?;
649
- let final_query = Self :: add_where_conditions ( base_query, aggregate_config, stream) . await ?;
654
+ let aggregate_function = Self :: parse_aggregate_function ( aggregate_config, alert_info) ?;
655
+ let base_query =
656
+ Self :: build_base_query ( & aggregate_function, aggregate_config, stream, alert_info) ?;
657
+ let final_query =
658
+ Self :: add_where_conditions ( base_query, aggregate_config, stream, alert_info) . await ?;
650
659
651
660
Ok ( final_query)
652
661
}
653
662
654
663
/// Parse aggregate function from v1 config
655
664
fn parse_aggregate_function (
656
665
aggregate_config : & JsonValue ,
666
+ alert_info : & str ,
657
667
) -> Result < AggregateFunction , AlertError > {
658
668
let aggregate_function_str =
659
669
aggregate_config[ "aggregateFunction" ]
660
670
. as_str ( )
661
671
. ok_or_else ( || {
662
- AlertError :: CustomError ( "Missing aggregateFunction in v1 alert" . to_string ( ) )
672
+ AlertError :: CustomError ( format ! (
673
+ "Missing aggregateFunction in v1 alert for {alert_info}"
674
+ ) )
663
675
} ) ?;
664
676
665
677
match aggregate_function_str. to_lowercase ( ) . as_str ( ) {
@@ -670,7 +682,7 @@ impl AlertConfig {
670
682
"max" => Ok ( AggregateFunction :: Max ) ,
671
683
"sum" => Ok ( AggregateFunction :: Sum ) ,
672
684
_ => Err ( AlertError :: CustomError ( format ! (
673
- "Unsupported aggregate function: {aggregate_function_str}" ,
685
+ "Unsupported aggregate function: {aggregate_function_str} for {alert_info}"
674
686
) ) ) ,
675
687
}
676
688
}
@@ -680,6 +692,7 @@ impl AlertConfig {
680
692
aggregate_function : & AggregateFunction ,
681
693
aggregate_config : & JsonValue ,
682
694
stream : & str ,
695
+ _alert_info : & str ,
683
696
) -> Result < String , AlertError > {
684
697
let column = aggregate_config[ "column" ] . as_str ( ) . unwrap_or ( "*" ) ;
685
698
@@ -716,6 +729,7 @@ impl AlertConfig {
716
729
base_query : String ,
717
730
aggregate_config : & JsonValue ,
718
731
stream : & str ,
732
+ alert_info : & str ,
719
733
) -> Result < String , AlertError > {
720
734
let Some ( conditions) = aggregate_config[ "conditions" ] . as_object ( ) else {
721
735
return Ok ( base_query) ;
@@ -734,7 +748,7 @@ impl AlertConfig {
734
748
Ok ( schema) => schema,
735
749
Err ( e) => {
736
750
return Err ( AlertError :: CustomError ( format ! (
737
- "Failed to fetch schema for stream '{stream}' during migration: {e}. Migration cannot proceed without schema information." ,
751
+ "Failed to fetch schema for stream '{stream}' during migration of {alert_info} : {e}. Migration cannot proceed without schema information."
738
752
) ) ) ;
739
753
}
740
754
} ;
@@ -743,15 +757,16 @@ impl AlertConfig {
743
757
for condition in condition_config {
744
758
let column = condition[ "column" ] . as_str ( ) . unwrap_or ( "" ) ;
745
759
if column. is_empty ( ) {
746
- warn ! ( "Skipping WHERE condition with empty column name" ) ;
760
+ warn ! ( "Skipping WHERE condition with empty column name for {alert_info} " ) ;
747
761
continue ;
748
762
}
749
763
let operator_str = condition[ "operator" ] . as_str ( ) . unwrap_or ( "=" ) ;
750
764
let value = condition[ "value" ] . as_str ( ) . unwrap_or ( "" ) ;
751
765
752
766
let operator = Self :: parse_where_operator ( operator_str) ;
753
- let where_clause =
754
- Self :: format_where_clause_with_types ( column, & operator, value, & schema) ?;
767
+ let where_clause = Self :: format_where_clause_with_types (
768
+ column, & operator, value, & schema, alert_info,
769
+ ) ?;
755
770
where_clauses. push ( where_clause) ;
756
771
}
757
772
@@ -789,6 +804,7 @@ impl AlertConfig {
789
804
operator : & WhereConfigOperator ,
790
805
value : & str ,
791
806
schema : & Schema ,
807
+ alert_info : & str ,
792
808
) -> Result < String , AlertError > {
793
809
match operator {
794
810
WhereConfigOperator :: IsNull | WhereConfigOperator :: IsNotNull => {
@@ -824,7 +840,8 @@ impl AlertConfig {
824
840
) ) ,
825
841
_ => {
826
842
// Standard operators: =, !=, <, >, <=, >=
827
- let formatted_value = Self :: convert_value_by_data_type ( column, value, schema) ?;
843
+ let formatted_value =
844
+ Self :: convert_value_by_data_type ( column, value, schema, alert_info) ?;
828
845
Ok ( format ! (
829
846
"\" {column}\" {} {formatted_value}" ,
830
847
operator. as_str( )
@@ -838,13 +855,14 @@ impl AlertConfig {
838
855
column : & str ,
839
856
value : & str ,
840
857
schema : & Schema ,
858
+ alert_info : & str ,
841
859
) -> Result < String , AlertError > {
842
860
// Find the field in the schema
843
861
let field = schema. fields ( ) . iter ( ) . find ( |f| f. name ( ) == column) ;
844
862
let Some ( field) = field else {
845
863
// Column not found in schema, fail migration
846
864
return Err ( AlertError :: CustomError ( format ! (
847
- "Column '{column}' not found in stream schema during migration. Available columns: [{}]" ,
865
+ "Column '{column}' not found in stream schema during migration of {alert_info} . Available columns: [{}]" ,
848
866
schema
849
867
. fields( )
850
868
. iter( )
@@ -859,23 +877,23 @@ impl AlertConfig {
859
877
match value. parse :: < f64 > ( ) {
860
878
Ok ( float_val) => Ok ( float_val. to_string ( ) ) , // Raw number without quotes
861
879
Err ( _) => Err ( AlertError :: CustomError ( format ! (
862
- "Failed to parse value '{value}' as float64 for column '{column}' during migration" ,
880
+ "Failed to parse value '{value}' as float64 for column '{column}' during migration of {alert_info} " ,
863
881
) ) ) ,
864
882
}
865
883
}
866
884
DataType :: Int64 => {
867
885
match value. parse :: < i64 > ( ) {
868
886
Ok ( int_val) => Ok ( int_val. to_string ( ) ) , // Raw number without quotes
869
887
Err ( _) => Err ( AlertError :: CustomError ( format ! (
870
- "Failed to parse value '{value}' as int64 for column '{column}' during migration" ,
888
+ "Failed to parse value '{value}' as int64 for column '{column}' during migration of {alert_info} " ,
871
889
) ) ) ,
872
890
}
873
891
}
874
892
DataType :: Boolean => {
875
893
match value. to_lowercase ( ) . parse :: < bool > ( ) {
876
894
Ok ( bool_val) => Ok ( bool_val. to_string ( ) ) , // Raw boolean without quotes
877
895
Err ( _) => Err ( AlertError :: CustomError ( format ! (
878
- "Failed to parse value '{value}' as boolean for column '{column}' during migration" ,
896
+ "Failed to parse value '{value}' as boolean for column '{column}' during migration of {alert_info} " ,
879
897
) ) ) ,
880
898
}
881
899
}
@@ -888,7 +906,7 @@ impl AlertConfig {
888
906
match value. parse :: < chrono:: DateTime < chrono:: Utc > > ( ) {
889
907
Ok ( _) => Ok ( format ! ( "'{}'" , value. replace( '\'' , "''" ) ) ) ,
890
908
Err ( _) => Err ( AlertError :: CustomError ( format ! (
891
- "Failed to parse value '{value}' as date for column '{column}' during migration" ,
909
+ "Failed to parse value '{value}' as date for column '{column}' during migration of {alert_info} " ,
892
910
) ) ) ,
893
911
}
894
912
}
@@ -899,7 +917,7 @@ impl AlertConfig {
899
917
match value. parse :: < chrono:: DateTime < chrono:: Utc > > ( ) {
900
918
Ok ( _) => Ok ( format ! ( "'{}'" , value. replace( '\'' , "''" ) ) ) ,
901
919
Err ( _) => Err ( AlertError :: CustomError ( format ! (
902
- "Failed to parse value '{value}' as timestamp for column '{column}' during migration" ,
920
+ "Failed to parse value '{value}' as timestamp for column '{column}' during migration of {alert_info} " ,
903
921
) ) ) ,
904
922
}
905
923
}
@@ -911,17 +929,20 @@ impl AlertConfig {
911
929
}
912
930
913
931
/// Extract threshold configuration from v1 alert
914
- fn extract_threshold_config ( alert_json : & JsonValue ) -> Result < ThresholdConfig , AlertError > {
932
+ fn extract_threshold_config (
933
+ alert_json : & JsonValue ,
934
+ alert_info : & str ,
935
+ ) -> Result < ThresholdConfig , AlertError > {
915
936
let aggregates = & alert_json[ "aggregates" ] ;
916
937
let aggregate_config = & aggregates[ "aggregateConfig" ] [ 0 ] ;
917
938
918
- let threshold_operator = aggregate_config[ "operator" ]
919
- . as_str ( )
920
- . ok_or_else ( || AlertError :: CustomError ( "Missing operator in v1 alert" . to_string ( ) ) ) ?;
939
+ let threshold_operator = aggregate_config[ "operator" ] . as_str ( ) . ok_or_else ( || {
940
+ AlertError :: CustomError ( format ! ( "Missing operator in v1 alert for {alert_info}" ) )
941
+ } ) ?;
921
942
922
- let threshold_value = aggregate_config[ "value" ]
923
- . as_f64 ( )
924
- . ok_or_else ( || AlertError :: CustomError ( "Missing value in v1 alert" . to_string ( ) ) ) ?;
943
+ let threshold_value = aggregate_config[ "value" ] . as_f64 ( ) . ok_or_else ( || {
944
+ AlertError :: CustomError ( format ! ( "Missing value in v1 alert for {alert_info}" ) )
945
+ } ) ?;
925
946
926
947
let operator = match threshold_operator {
927
948
">" => AlertOperator :: GreaterThan ,
@@ -940,21 +961,30 @@ impl AlertConfig {
940
961
}
941
962
942
963
/// Extract evaluation configuration from v1 alert
943
- fn extract_eval_config ( alert_json : & JsonValue ) -> Result < EvalConfig , AlertError > {
964
+ fn extract_eval_config (
965
+ alert_json : & JsonValue ,
966
+ alert_info : & str ,
967
+ ) -> Result < EvalConfig , AlertError > {
944
968
let rolling_window = & alert_json[ "evalConfig" ] [ "rollingWindow" ] ;
945
969
946
970
let eval_start = rolling_window[ "evalStart" ]
947
971
. as_str ( )
948
- . ok_or_else ( || AlertError :: CustomError ( "Missing evalStart in v1 alert" . to_string ( ) ) ) ?
972
+ . ok_or_else ( || {
973
+ AlertError :: CustomError ( format ! ( "Missing evalStart in v1 alert for {alert_info}" ) )
974
+ } ) ?
949
975
. to_string ( ) ;
950
976
951
977
let eval_end = rolling_window[ "evalEnd" ]
952
978
. as_str ( )
953
- . ok_or_else ( || AlertError :: CustomError ( "Missing evalEnd in v1 alert" . to_string ( ) ) ) ?
979
+ . ok_or_else ( || {
980
+ AlertError :: CustomError ( format ! ( "Missing evalEnd in v1 alert for {alert_info}" ) )
981
+ } ) ?
954
982
. to_string ( ) ;
955
983
956
984
let eval_frequency = rolling_window[ "evalFrequency" ] . as_u64 ( ) . ok_or_else ( || {
957
- AlertError :: CustomError ( "Missing evalFrequency in v1 alert" . to_string ( ) )
985
+ AlertError :: CustomError ( format ! (
986
+ "Missing evalFrequency in v1 alert for {alert_info}"
987
+ ) )
958
988
} ) ?;
959
989
960
990
Ok ( EvalConfig :: RollingWindow ( RollingWindow {
@@ -965,16 +995,24 @@ impl AlertConfig {
965
995
}
966
996
967
997
/// Extract target IDs from v1 alert
968
- fn extract_targets ( alert_json : & JsonValue ) -> Result < Vec < Ulid > , AlertError > {
998
+ fn extract_targets ( alert_json : & JsonValue , alert_info : & str ) -> Result < Vec < Ulid > , AlertError > {
969
999
let targets: Result < Vec < Ulid > , _ > = alert_json[ "targets" ]
970
1000
. as_array ( )
971
- . ok_or_else ( || AlertError :: CustomError ( "Missing targets in v1 alert" . to_string ( ) ) ) ?
1001
+ . ok_or_else ( || {
1002
+ AlertError :: CustomError ( format ! ( "Missing targets in v1 alert for {alert_info}" ) )
1003
+ } ) ?
972
1004
. iter ( )
973
1005
. map ( |t| {
974
1006
t. as_str ( )
975
- . ok_or_else ( || AlertError :: CustomError ( "Invalid target format" . to_string ( ) ) ) ?
1007
+ . ok_or_else ( || {
1008
+ AlertError :: CustomError ( format ! ( "Invalid target format for {alert_info}" ) )
1009
+ } ) ?
976
1010
. parse ( )
977
- . map_err ( |_| AlertError :: CustomError ( "Invalid target ID format" . to_string ( ) ) )
1011
+ . map_err ( |_| {
1012
+ AlertError :: CustomError ( format ! (
1013
+ "Invalid target ID format for {alert_info}"
1014
+ ) )
1015
+ } )
978
1016
} )
979
1017
. collect ( ) ;
980
1018
0 commit comments