@@ -343,14 +343,30 @@ pub fn filter_metric_tags(
343343
344344#[ cfg( test) ]
345345mod tests {
346+ use std:: sync:: Arc ;
347+
346348 use saluki_config:: { dynamic:: ConfigUpdate , ConfigurationLoader } ;
347349 use saluki_context:: {
348350 tags:: { Tag , TagSet } ,
349351 Context , TagSetMutViewState ,
350352 } ;
351- use saluki_core:: data_model:: event:: metric:: Metric ;
353+ use saluki_core:: accounting:: { ComponentRegistry , MemoryLimiter } ;
354+ use saluki_core:: components:: {
355+ transforms:: { TransformBuilder , TransformContext } ,
356+ ComponentContext ,
357+ } ;
358+ use saluki_core:: data_model:: event:: {
359+ metric:: { Metric , MetricValues } ,
360+ Event ,
361+ } ;
362+ use saluki_core:: health:: HealthRegistry ;
363+ use saluki_core:: runtime:: { state:: DataspaceRegistry , Supervisor } ;
364+ use saluki_core:: topology:: interconnect:: { Consumer , Dispatcher } ;
365+ use saluki_core:: topology:: { EventsBuffer , OutputName , TopologyContext } ;
352366 use saluki_metrics:: { test:: TestRecorder , MetricsBuilder } ;
353367 use serde_json:: json;
368+ use tokio:: runtime:: Handle ;
369+ use tokio:: sync:: mpsc;
354370
355371 use super :: * ;
356372
@@ -1038,4 +1054,127 @@ mod tests {
10381054 assert ! ( !metric. context( ) . tags( ) . is_modified( ) ) ;
10391055 assert ! ( !metric. context( ) . origin_tags( ) . is_modified( ) ) ;
10401056 }
1057+
1058+ #[ tokio:: test]
1059+ async fn run_loop_enforces_type_guard_and_exercises_context_cache ( ) {
1060+ // The other tests call `filter_metric_tags` directly; this one drives the real `Transform::run()` loop to
1061+ // cover two behaviors those can't reach:
1062+ // 1. the run-loop type guard filters only distribution (sketch) and count metrics, leaving other metric
1063+ // types (a gauge here) completely untouched even when a rule matches their name; and
1064+ // 2. the per-context dedup cache is exercised end-to-end -- the cache is keyed by `Context`, so metrics
1065+ // that share a (name, tags) context resolve to a single cache entry, and every metric sharing that
1066+ // context is filtered identically (the second and later occurrences take the cache-hit branch).
1067+
1068+ let cfg_json = json ! ( {
1069+ "metric_tag_filterlist" : [
1070+ { "metric_name" : "svc.latency" , "action" : "exclude" , "tags" : [ "host" ] }
1071+ ]
1072+ } ) ;
1073+ let ( config, _sender) = ConfigurationLoader :: for_tests ( Some ( cfg_json) , None , false ) . await ;
1074+ let builder = TagFilterlistConfiguration :: from_configuration ( & config) . expect ( "config should parse" ) ;
1075+
1076+ let component_context = ComponentContext :: test_transform ( "tag_filterlist" ) ;
1077+ let transform = builder
1078+ . build ( component_context. clone ( ) )
1079+ . await
1080+ . expect ( "tag filterlist should build" ) ;
1081+
1082+ // Wire a dispatcher whose default output we can drain after the run loop completes.
1083+ let mut dispatcher = Dispatcher :: new ( component_context. clone ( ) ) ;
1084+ dispatcher. add_output ( OutputName :: Default ) . expect ( "add default output" ) ;
1085+ let ( out_tx, mut out_rx) = mpsc:: channel ( 4 ) ;
1086+ dispatcher
1087+ . attach_sender_to_output ( & OutputName :: Default , out_tx)
1088+ . expect ( "attach default sender" ) ;
1089+
1090+ // A distribution, a counter, and a gauge that all share the same (name, tags) context, followed by a repeat
1091+ // of the distribution. The counter and the repeated distribution hit the cache entry created by the first
1092+ // distribution.
1093+ let tags = & [ "host:h1" , "env:prod" ] ;
1094+ let mut input = EventsBuffer :: default ( ) ;
1095+ for event in [
1096+ Event :: Metric ( Metric :: distribution (
1097+ Context :: from_static_parts ( "svc.latency" , tags) ,
1098+ 1.0 ,
1099+ ) ) ,
1100+ Event :: Metric ( Metric :: counter ( Context :: from_static_parts ( "svc.latency" , tags) , 1.0 ) ) ,
1101+ Event :: Metric ( Metric :: gauge ( Context :: from_static_parts ( "svc.latency" , tags) , 1.0 ) ) ,
1102+ Event :: Metric ( Metric :: distribution (
1103+ Context :: from_static_parts ( "svc.latency" , tags) ,
1104+ 1.0 ,
1105+ ) ) ,
1106+ ] {
1107+ assert ! ( input. try_push( event) . is_none( ) , "input buffer should have capacity" ) ;
1108+ }
1109+
1110+ let ( in_tx, in_rx) = mpsc:: channel ( 4 ) ;
1111+ let consumer = Consumer :: new ( component_context. clone ( ) , in_rx) ;
1112+ in_tx. send ( input) . await . expect ( "send input buffer" ) ;
1113+ drop ( in_tx) ; // Closing the input makes the run loop terminate deterministically.
1114+
1115+ let topology_context = TopologyContext :: new (
1116+ Arc :: from ( "test" ) ,
1117+ MemoryLimiter :: noop ( ) ,
1118+ HealthRegistry :: new ( ) ,
1119+ Handle :: current ( ) ,
1120+ DataspaceRegistry :: new ( ) ,
1121+ ) ;
1122+ let health = HealthRegistry :: new ( )
1123+ . register_component ( & saluki_core:: support:: SubsystemIdentifier :: from_dotted ( "test" ) )
1124+ . expect ( "component was not previously registered" ) ;
1125+ let supervisor_handle = Supervisor :: new ( "test" ) . expect ( "valid supervisor name" ) . handle ( ) ;
1126+
1127+ let context = TransformContext :: new (
1128+ & topology_context,
1129+ & component_context,
1130+ ComponentRegistry :: default ( ) ,
1131+ health,
1132+ dispatcher,
1133+ consumer,
1134+ supervisor_handle,
1135+ ) ;
1136+
1137+ transform. run ( context) . await . expect ( "tag filterlist run should succeed" ) ;
1138+
1139+ let mut dispatched: Vec < Metric > = Vec :: new ( ) ;
1140+ while let Ok ( buffer) = out_rx. try_recv ( ) {
1141+ for event in buffer {
1142+ if let Event :: Metric ( metric) = event {
1143+ dispatched. push ( metric) ;
1144+ }
1145+ }
1146+ }
1147+
1148+ // Nothing is dropped by the transform; order is preserved.
1149+ assert_eq ! ( dispatched. len( ) , 4 , "all four metrics should be dispatched" ) ;
1150+
1151+ let sorted_tags = |metric : & Metric | {
1152+ let mut names: Vec < String > = metric
1153+ . context ( )
1154+ . tags ( )
1155+ . into_iter ( )
1156+ . map ( |t| t. as_str ( ) . to_owned ( ) )
1157+ . collect ( ) ;
1158+ names. sort ( ) ;
1159+ names
1160+ } ;
1161+
1162+ // Distribution (sketch) -> filtered on the cache-miss path.
1163+ assert ! ( dispatched[ 0 ] . values( ) . is_sketch( ) ) ;
1164+ assert_eq ! ( sorted_tags( & dispatched[ 0 ] ) , vec![ "env:prod" ] ) ;
1165+ // Counter (count metric) -> filtered via the cache-hit branch (shares the distribution's context entry).
1166+ assert ! ( matches!( dispatched[ 1 ] . values( ) , MetricValues :: Counter ( _) ) ) ;
1167+ assert_eq ! ( sorted_tags( & dispatched[ 1 ] ) , vec![ "env:prod" ] ) ;
1168+ // Gauge -> NOT a sketch and NOT a counter, so the type guard skips it and it passes through untouched.
1169+ assert ! ( !dispatched[ 2 ] . values( ) . is_sketch( ) ) ;
1170+ assert ! ( !matches!( dispatched[ 2 ] . values( ) , MetricValues :: Counter ( _) ) ) ;
1171+ assert_eq ! (
1172+ sorted_tags( & dispatched[ 2 ] ) ,
1173+ vec![ "env:prod" , "host:h1" ] ,
1174+ "gauge metrics must not be filtered by the type guard"
1175+ ) ;
1176+ // Repeated distribution -> filtered via the cache-hit branch, identical to the first distribution.
1177+ assert ! ( dispatched[ 3 ] . values( ) . is_sketch( ) ) ;
1178+ assert_eq ! ( sorted_tags( & dispatched[ 3 ] ) , vec![ "env:prod" ] ) ;
1179+ }
10411180}
0 commit comments