1717 */
1818
1919use actix_web:: http:: header:: HeaderMap ;
20+ use datafusion:: common:: HashSet ;
21+ use ulid:: Ulid ;
2022
2123use crate :: {
22- event:: format:: LogSource ,
23- handlers:: {
24+ event:: format:: LogSource , handlers:: {
2425 CUSTOM_PARTITION_KEY , LOG_SOURCE_KEY , STATIC_SCHEMA_FLAG , STREAM_TYPE_KEY ,
2526 TELEMETRY_TYPE_KEY , TIME_PARTITION_KEY , TIME_PARTITION_LIMIT_KEY , TelemetryType ,
2627 UPDATE_STREAM_KEY ,
27- } ,
28- storage:: StreamType ,
28+ } ,
29+ metastore:: MetastoreError ,
30+ parseable:: {
31+ PARSEABLE , StreamNotFound
32+ } ,
33+ storage:: StreamType ,
34+ users:: {
35+ dashboards:: { Dashboard } ,
36+ filters:: Filter
37+ }
2938} ;
3039
40+ /// Field in a dashboard's tile that should contain the logstream name
41+ const TILE_FIELD_REFERRING_TO_STREAM : & str = "chartQuery" ;
42+
3143#[ derive( Debug , Default ) ]
3244pub struct PutStreamHeaders {
3345 pub time_partition : String ,
@@ -74,3 +86,138 @@ impl From<&HeaderMap> for PutStreamHeaders {
7486 }
7587 }
7688}
89+
90+ /// Resources that rely on a specific logstream and will be affected if it gets deleted
91+ #[ derive( Debug , Default , serde:: Serialize ) ]
92+ pub struct LogstreamAffectedResources {
93+ pub filters : Vec < Filter > ,
94+ pub dashboards : Vec < LogstreamAffectedDashboard >
95+ }
96+
97+ #[ derive( Debug , Default , serde:: Serialize ) ]
98+ pub struct LogstreamAffectedDashboard {
99+ pub dashboard : Dashboard ,
100+ pub affected_tile_ids : Vec < Ulid >
101+ }
102+
103+ #[ derive( thiserror:: Error , Debug ) ]
104+ pub enum LogstreamAffectedResourcesError {
105+ #[ error( "Stream not found: {0}" ) ]
106+ NoSuchStream ( #[ from] StreamNotFound ) ,
107+
108+ #[ error( "Metastore error: {0}" ) ]
109+ FromMetastoreError ( #[ from] MetastoreError ) ,
110+ }
111+
112+ impl LogstreamAffectedResources {
113+ pub async fn load ( stream_name : & str ) -> Self {
114+ Self {
115+ filters : LogstreamAffectedResources :: fetch_affected_filters ( stream_name)
116+ . await
117+ . unwrap_or_else ( |e| {
118+ tracing:: warn!( "failed to fetch filters: {}" , e) ;
119+ Vec :: new ( )
120+ } ) ,
121+
122+ dashboards : LogstreamAffectedResources :: fetch_affected_dashboards ( stream_name)
123+ . await
124+ . unwrap_or_else ( |e| {
125+ tracing:: warn!( "failed to fetch dashboards: {}" , e) ;
126+ Vec :: new ( )
127+ } ) ,
128+ }
129+ }
130+
131+ pub async fn fetch_affected_filters (
132+ stream_name : & str
133+ ) -> Result < Vec < Filter > , LogstreamAffectedResourcesError > {
134+ if !PARSEABLE . streams . contains ( stream_name) {
135+ return Err ( LogstreamAffectedResourcesError :: NoSuchStream (
136+ StreamNotFound ( stream_name. to_string ( ) )
137+ ) ) ;
138+ }
139+
140+ Ok ( PARSEABLE . metastore . get_filters ( ) . await ?
141+ . into_iter ( )
142+ . filter ( |filter| filter. stream_name == stream_name)
143+ . collect ( ) )
144+ }
145+
146+ pub async fn fetch_affected_dashboards (
147+ stream_name : & str
148+ ) -> Result < Vec < LogstreamAffectedDashboard > , LogstreamAffectedResourcesError > {
149+ if !PARSEABLE . streams . contains ( stream_name) {
150+ return Err ( LogstreamAffectedResourcesError :: NoSuchStream (
151+ StreamNotFound ( stream_name. to_string ( ) )
152+ ) ) ;
153+ }
154+
155+ let all_dashboards = PARSEABLE . metastore . get_dashboards ( ) . await ?;
156+ let mut parsed_dashboards = Vec :: < Dashboard > :: new ( ) ;
157+
158+ for dashboard in all_dashboards {
159+ if dashboard. is_empty ( ) {
160+ continue ;
161+ }
162+
163+ let dashboard_value = match serde_json:: from_slice :: < serde_json:: Value > ( & dashboard) {
164+ Ok ( value) => value,
165+ Err ( err) => {
166+ tracing:: warn!( "Failed to parse dashboard JSON: {}" , err) ;
167+ continue ;
168+ }
169+ } ;
170+
171+ if let Ok ( dashboard) = serde_json:: from_value :: < Dashboard > ( dashboard_value. clone ( ) ) {
172+ parsed_dashboards. retain ( |d : & Dashboard | {
173+ d. dashboard_id != dashboard. dashboard_id
174+ } ) ;
175+
176+ parsed_dashboards. push ( dashboard) ;
177+ } else {
178+ tracing:: warn!( "Failed to deserialize dashboard: {:?}" , dashboard_value) ;
179+ }
180+ }
181+
182+ let mut affected_dashboards: Vec < LogstreamAffectedDashboard > = vec ! [ ] ;
183+
184+ for dashboard in parsed_dashboards {
185+ let Some ( tiles) = dashboard. tiles . as_ref ( ) else {
186+ continue ;
187+ } ;
188+
189+ println ! ( "here" ) ;
190+
191+ let mut affected_tile_ids = HashSet :: < Ulid > :: new ( ) ;
192+
193+ for tile in tiles {
194+ let Some ( tile_fields) = tile. other_fields . as_ref ( ) else {
195+ continue ;
196+ } ;
197+
198+ let Some ( tile_value) = tile_fields. get ( TILE_FIELD_REFERRING_TO_STREAM ) else {
199+ continue ;
200+ } ;
201+
202+
203+
204+ if let Some ( chart_query) = tile_value. as_str ( ) {
205+ println ! ( "{}" , chart_query) ;
206+ if chart_query. contains ( stream_name) && !affected_tile_ids. contains ( & tile. tile_id ) {
207+ affected_tile_ids. insert ( tile. tile_id ) ;
208+ }
209+ }
210+ }
211+
212+ if !affected_tile_ids. is_empty ( ) {
213+ affected_dashboards. push ( LogstreamAffectedDashboard {
214+ dashboard,
215+ affected_tile_ids : affected_tile_ids. into_iter ( ) . collect ( )
216+ } ) ;
217+ }
218+ }
219+
220+ println ! ( "h2: {}" , affected_dashboards. len( ) ) ;
221+ Ok ( affected_dashboards)
222+ }
223+ }
0 commit comments