From 4e5e0b9847ed76d7a638e64fc60cd39d06fc4d9a Mon Sep 17 00:00:00 2001 From: omercengiz Date: Fri, 4 Sep 2026 23:47:42 +0300 Subject: [PATCH] feat: make structural analysis configurable --- README.md | 7 +++ src/flowsense/application/analyzer.py | 63 +++++++++++++++++++-------- src/flowsense/cli/main.py | 26 +++++++++++ src/flowsense/domain/policy.py | 17 ++++++++ src/flowsense/mcp/server.py | 33 ++++++++++++++ tests/test_analyzer.py | 29 ++++++++++++ tests/test_cli.py | 39 ++++++++++++++++- tests/test_mcp_server.py | 15 +++++++ tests/test_policy.py | 18 ++++++++ 9 files changed, 229 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 2142f6c..7dd3471 100644 --- a/README.md +++ b/README.md @@ -177,12 +177,19 @@ policy = AnalysisPolicy( high_threshold=4.0, critical_threshold=6.0, mapped_task_aggregation=MappedTaskAggregation.MAX, + change_point_minimum_segment_size=4, + change_point_score_threshold=4.0, + trend_minimum_observations=8, + trend_score_threshold=4.0, + trend_minimum_directional_consistency=0.75, ) ``` `baseline_window` limits the number of historical values used before the current run. Mapped task durations can be aggregated with `MAX`, `MEAN`, or `SUM`. The same policy options are available through the CLI and MCP tool. +Change-point and trend detection can also be disabled independently with +`change_point_detection_enabled=False` or `trend_detection_enabled=False`. Every `DAGAnalysis` exposes a derived `summary` with task-analysis coverage, severity distribution, anomalous task and handoff counts, uniquely affected diff --git a/src/flowsense/application/analyzer.py b/src/flowsense/application/analyzer.py index 7e55eb1..0e941fb 100644 --- a/src/flowsense/application/analyzer.py +++ b/src/flowsense/application/analyzer.py @@ -40,13 +40,28 @@ def analyze_dag( drift_results = {} for task_id, durations in duration_history.items(): - change_point = detect_change_point(task_id, durations) - if change_point is not None: - change_point_results[task_id] = change_point - - trend = detect_trend(task_id, durations) - if trend is not None: - trend_results[task_id] = trend + if policy.change_point_detection_enabled: + change_point = detect_change_point( + task_id, + durations, + minimum_segment_size=policy.change_point_minimum_segment_size, + score_threshold=policy.change_point_score_threshold, + ) + if change_point is not None: + change_point_results[task_id] = change_point + + if policy.trend_detection_enabled: + trend = detect_trend( + task_id, + durations, + minimum_observations=policy.trend_minimum_observations, + score_threshold=policy.trend_score_threshold, + minimum_directional_consistency=( + policy.trend_minimum_directional_consistency + ), + ) + if trend is not None: + trend_results[task_id] = trend try: drift_results[task_id] = calculate_drift( @@ -86,16 +101,30 @@ def analyze_dag( for edge, delays in handoff_history.items(): upstream_task, downstream_task = edge - change_point = detect_change_point( - f"{upstream_task}->{downstream_task}", - delays, - ) - if change_point is not None: - handoff_change_point_results[edge] = change_point - - trend = detect_trend(f"{upstream_task}->{downstream_task}", delays) - if trend is not None: - handoff_trend_results[edge] = trend + subject_id = f"{upstream_task}->{downstream_task}" + + if policy.change_point_detection_enabled: + change_point = detect_change_point( + subject_id, + delays, + minimum_segment_size=policy.change_point_minimum_segment_size, + score_threshold=policy.change_point_score_threshold, + ) + if change_point is not None: + handoff_change_point_results[edge] = change_point + + if policy.trend_detection_enabled: + trend = detect_trend( + subject_id, + delays, + minimum_observations=policy.trend_minimum_observations, + score_threshold=policy.trend_score_threshold, + minimum_directional_consistency=( + policy.trend_minimum_directional_consistency + ), + ) + if trend is not None: + handoff_trend_results[edge] = trend try: handoff_drift_results[edge] = calculate_handoff_drift( diff --git a/src/flowsense/cli/main.py b/src/flowsense/cli/main.py index 071ff9e..19a882d 100644 --- a/src/flowsense/cli/main.py +++ b/src/flowsense/cli/main.py @@ -35,6 +35,23 @@ def analyze( medium_threshold: float = typer.Option(2.0, min=0.0), high_threshold: float = typer.Option(3.5, min=0.0), critical_threshold: float = typer.Option(5.0, min=0.0), + change_point_detection: bool = typer.Option( + True, + "--change-point-detection/--no-change-point-detection", + ), + change_point_minimum_segment_size: int = typer.Option(3, min=2), + change_point_score_threshold: float = typer.Option(3.5, min=0.0), + trend_detection: bool = typer.Option( + True, + "--trend-detection/--no-trend-detection", + ), + trend_minimum_observations: int = typer.Option(5, min=3), + trend_score_threshold: float = typer.Option(3.5, min=0.0), + trend_minimum_directional_consistency: float = typer.Option( + 0.6, + min=0.0, + max=1.0, + ), mapped_task_aggregation: Annotated[ MappedTaskAggregation, typer.Option(), @@ -48,6 +65,15 @@ def analyze( high_threshold=high_threshold, critical_threshold=critical_threshold, mapped_task_aggregation=mapped_task_aggregation, + change_point_detection_enabled=change_point_detection, + change_point_minimum_segment_size=change_point_minimum_segment_size, + change_point_score_threshold=change_point_score_threshold, + trend_detection_enabled=trend_detection, + trend_minimum_observations=trend_minimum_observations, + trend_score_threshold=trend_score_threshold, + trend_minimum_directional_consistency=( + trend_minimum_directional_consistency + ), ) except ValueError as exc: raise typer.BadParameter(str(exc)) from exc diff --git a/src/flowsense/domain/policy.py b/src/flowsense/domain/policy.py index e24d710..84eecad 100644 --- a/src/flowsense/domain/policy.py +++ b/src/flowsense/domain/policy.py @@ -11,6 +11,13 @@ class AnalysisPolicy: high_threshold: float = 3.5 critical_threshold: float = 5.0 mapped_task_aggregation: MappedTaskAggregation = MappedTaskAggregation.MAX + change_point_detection_enabled: bool = True + change_point_minimum_segment_size: int = 3 + change_point_score_threshold: float = 3.5 + trend_detection_enabled: bool = True + trend_minimum_observations: int = 5 + trend_score_threshold: float = 3.5 + trend_minimum_directional_consistency: float = 0.6 def __post_init__(self) -> None: if self.minimum_history < 2: @@ -26,6 +33,16 @@ def __post_init__(self) -> None: 0 < self.medium_threshold < self.high_threshold < self.critical_threshold ): raise ValueError("Severity thresholds must be positive and increasing.") + if self.change_point_minimum_segment_size < 2: + raise ValueError("change_point_minimum_segment_size must be at least 2.") + if self.change_point_score_threshold <= 0: + raise ValueError("change_point_score_threshold must be positive.") + if self.trend_minimum_observations < 3: + raise ValueError("trend_minimum_observations must be at least 3.") + if self.trend_score_threshold <= 0: + raise ValueError("trend_score_threshold must be positive.") + if not 0 < self.trend_minimum_directional_consistency <= 1: + raise ValueError("trend_minimum_directional_consistency must be in (0, 1].") DEFAULT_ANALYSIS_POLICY = AnalysisPolicy() diff --git a/src/flowsense/mcp/server.py b/src/flowsense/mcp/server.py index 94d8d84..6faf9d3 100644 --- a/src/flowsense/mcp/server.py +++ b/src/flowsense/mcp/server.py @@ -42,6 +42,21 @@ def serialize_analysis(analysis: DAGAnalysis) -> dict: "high_threshold": analysis.policy.high_threshold, "critical_threshold": analysis.policy.critical_threshold, "mapped_task_aggregation": analysis.policy.mapped_task_aggregation, + "change_point_detection_enabled": ( + analysis.policy.change_point_detection_enabled + ), + "change_point_minimum_segment_size": ( + analysis.policy.change_point_minimum_segment_size + ), + "change_point_score_threshold": ( + analysis.policy.change_point_score_threshold + ), + "trend_detection_enabled": analysis.policy.trend_detection_enabled, + "trend_minimum_observations": (analysis.policy.trend_minimum_observations), + "trend_score_threshold": analysis.policy.trend_score_threshold, + "trend_minimum_directional_consistency": ( + analysis.policy.trend_minimum_directional_consistency + ), }, "primary_origin": ( { @@ -164,6 +179,13 @@ def analyze_airflow_dag( medium_threshold: float = 2.0, high_threshold: float = 3.5, critical_threshold: float = 5.0, + change_point_detection_enabled: bool = True, + change_point_minimum_segment_size: int = 3, + change_point_score_threshold: float = 3.5, + trend_detection_enabled: bool = True, + trend_minimum_observations: int = 5, + trend_score_threshold: float = 3.5, + trend_minimum_directional_consistency: float = 0.6, mapped_task_aggregation: MappedTaskAggregation = MappedTaskAggregation.MAX, ) -> dict: """Analyze an Apache Airflow DAG for temporal drift and propagation.""" @@ -179,6 +201,17 @@ def analyze_airflow_dag( high_threshold=high_threshold, critical_threshold=critical_threshold, mapped_task_aggregation=mapped_task_aggregation, + change_point_detection_enabled=change_point_detection_enabled, + change_point_minimum_segment_size=( + change_point_minimum_segment_size + ), + change_point_score_threshold=change_point_score_threshold, + trend_detection_enabled=trend_detection_enabled, + trend_minimum_observations=trend_minimum_observations, + trend_score_threshold=trend_score_threshold, + trend_minimum_directional_consistency=( + trend_minimum_directional_consistency + ), ), ) except AirflowApiError as exc: diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py index edacba0..bfbb799 100644 --- a/tests/test_analyzer.py +++ b/tests/test_analyzer.py @@ -3,6 +3,7 @@ import pytest +from flowsense import AnalysisPolicy from flowsense.application import DAGDataSource, analyze_dag from flowsense.models import TaskRun @@ -185,6 +186,34 @@ def test_analyze_dag_detects_task_and_handoff_trends() -> None: assert handoff_trend.slope_per_observation == 1.0 +def test_analyze_dag_uses_structural_analysis_policy() -> None: + client = MagicMock(spec=DAGDataSource) + client.collect_task_runs.return_value = [ + TaskRun( + dag_id="demo", + dag_run_id=f"run_{index}", + task_id="transform", + state="success", + duration=duration, + ) + for index, duration in enumerate( + [3.0, 4.0, 5.0, 6.0, 7.0, 8.0], + start=1, + ) + ] + client.get_dag_dependencies.return_value = {"transform": []} + policy = AnalysisPolicy( + change_point_detection_enabled=False, + trend_detection_enabled=False, + ) + + analysis = analyze_dag("demo", client, policy=policy) + + assert analysis.change_point_results == {} + assert analysis.trend_results == {} + assert analysis.policy is policy + + def test_analyze_dag_calculates_handoff_drift() -> None: client = MagicMock(spec=DAGDataSource) diff --git a/tests/test_cli.py b/tests/test_cli.py index b6c8204..288a0d2 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,4 @@ -from unittest.mock import patch +from unittest.mock import MagicMock, patch from typer.testing import CliRunner @@ -20,3 +20,40 @@ def test_analyze_reports_airflow_api_errors() -> None: assert result.exit_code == 1 assert "Airflow request failed" in result.output assert "503" in result.output + + +def test_analyze_builds_structural_analysis_policy_from_options() -> None: + with ( + patch("flowsense.cli.main.AirflowClient"), + patch("flowsense.cli.main.analyze_dag", return_value=MagicMock()) as analyze, + patch("flowsense.cli.main.render_analysis"), + ): + result = CliRunner().invoke( + app, + [ + "analyze", + "demo", + "--no-change-point-detection", + "--change-point-minimum-segment-size", + "4", + "--change-point-score-threshold", + "4.5", + "--no-trend-detection", + "--trend-minimum-observations", + "8", + "--trend-score-threshold", + "4.5", + "--trend-minimum-directional-consistency", + "0.75", + ], + ) + + assert result.exit_code == 0 + policy = analyze.call_args.kwargs["policy"] + assert policy.change_point_detection_enabled is False + assert policy.change_point_minimum_segment_size == 4 + assert policy.change_point_score_threshold == 4.5 + assert policy.trend_detection_enabled is False + assert policy.trend_minimum_observations == 8 + assert policy.trend_score_threshold == 4.5 + assert policy.trend_minimum_directional_consistency == 0.75 diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 39d1356..6d0b93b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -156,6 +156,21 @@ def test_serialize_analysis() -> None: assert result["dag_id"] == "demo" assert result["runs_analyzed"] == 5 assert result["overall_severity"] == "CRITICAL" + assert result["policy"] == { + "minimum_history": 5, + "baseline_window": None, + "medium_threshold": 2.0, + "high_threshold": 3.5, + "critical_threshold": 5.0, + "mapped_task_aggregation": "MAX", + "change_point_detection_enabled": True, + "change_point_minimum_segment_size": 3, + "change_point_score_threshold": 3.5, + "trend_detection_enabled": True, + "trend_minimum_observations": 5, + "trend_score_threshold": 3.5, + "trend_minimum_directional_consistency": 0.6, + } assert result["summary"] == { "total_tasks": 2, "analyzed_tasks": 1, diff --git a/tests/test_policy.py b/tests/test_policy.py index 459af2a..7abbc4d 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -20,6 +20,24 @@ def test_policy_validates_history_and_thresholds() -> None: AnalysisPolicy(minimum_history=5, baseline_window=3) +@pytest.mark.parametrize( + "overrides", + [ + {"change_point_minimum_segment_size": 1}, + {"change_point_score_threshold": 0.0}, + {"trend_minimum_observations": 2}, + {"trend_score_threshold": 0.0}, + {"trend_minimum_directional_consistency": 0.0}, + {"trend_minimum_directional_consistency": 1.1}, + ], +) +def test_policy_validates_structural_analysis_settings( + overrides: dict[str, int | float], +) -> None: + with pytest.raises(ValueError): + AnalysisPolicy(**overrides) + + def test_drift_uses_configured_minimum_history_and_thresholds() -> None: policy = AnalysisPolicy( minimum_history=3,