Choreo: Advanced Architectural Analysis

Mix.install([
  {:choreo, path: "/home/mafinar/repos/elixir/choreo"},
  {:kino_vizjs, "~> 0.8.0"}
])

Section

> Deeper Insights: Beyond simple visualization, Choreo provides structural metrics to identify single points of failure, coupled nuclei, and performance hotspots.

1. Single Point of Failure (SPOF) Detection

An Articulation Point (or Cut Vertex) is a node that, if removed, would split the graph into two or more disconnected components. In a microservice architecture, these are your most critical nodes.

alias Choreo
alias Choreo.Analysis

system =
  Choreo.new()
  |> Choreo.add_service(:a, name: "Svc A")
  |> Choreo.add_service(:b, name: "Critical Hub")
  |> Choreo.add_service(:c, name: "Svc C")
  |> Choreo.add_service(:d, name: "Svc D")
  |> Choreo.connect(:a, :b)
  |> Choreo.connect(:b, :c)
  |> Choreo.connect(:b, :d)

# Highlight SPOFs in Red
spof_system = Analysis.heatmap(system, measure: :spof, palette: :heat)

Kino.Layout.grid([
  Kino.VizJS.render(Choreo.to_dot(spof_system)),
  Kino.VizJS.render(Choreo.to_dot(Analysis.legend(palette: :heat)))
], columns: 2)

2. Nucleus Detection (K-Core)

K-Core Decomposition identifies nested layers of connectivity. Higher “Core Numbers” represent nodes that are part of a tightly-coupled nucleus.

# A more complex mesh
mesh = Choreo.new()
  |> Choreo.add_service(:n1) |> Choreo.add_service(:n2) |> Choreo.add_service(:n3)
  |> Choreo.add_service(:n4) |> Choreo.add_service(:n5) |> Choreo.add_service(:n6)
  # The 3-core "Nucleus"
  |> Choreo.connect(:n1, :n2) |> Choreo.connect(:n2, :n3) |> Choreo.connect(:n3, :n1)
  |> Choreo.connect(:n1, :n4) |> Choreo.connect(:n2, :n4) |> Choreo.connect(:n3, :n4)
  # Peripheral nodes
  |> Choreo.add_service(:p1) |> Choreo.connect(:n4, :p1)

# Visualize the "Core"
# The nucleus (n1, n2, n3, n4) will be red/orange, peripheral p1 will be pale yellow.
core_system = Analysis.heatmap(mesh, measure: :k_core, palette: :cool)

Kino.Layout.grid([
  Kino.VizJS.render(Choreo.to_dot(core_system)),
  Kino.VizJS.render(Choreo.to_dot(Analysis.legend(palette: :cool)))
], columns: 2)

3. Workflow Latency Hotspots

Workflow heatmaps highlight tasks based on their cumulative latency (including retries).

alias Choreo.Workflow
alias Choreo.Workflow.Analysis, as: WorkflowAnalysis

wf = Workflow.new()
  |> Workflow.add_start(:start)
  |> Workflow.add_task(:fast, timeout_ms: 100)
  |> Workflow.add_task(:slow, timeout_ms: 5000)
  |> Workflow.add_end(:done)
  |> Workflow.connect(:start, :fast)
  |> Workflow.connect(:fast, :slow)
  |> Workflow.connect(:slow, :done)

# Highlight slow tasks
heat_wf = WorkflowAnalysis.heatmap(wf, palette: :spectral)

Kino.Layout.grid([
  Kino.VizJS.render(Workflow.to_dot(heat_wf)),
  Kino.VizJS.render(Choreo.to_dot(Analysis.legend(palette: :spectral)))
], columns: 2)

4. Dataflow Throughput Hotspots

Dataflow heatmaps visualize volume by coloring stages based on their inbound event rate.

alias Choreo.Dataflow
alias Choreo.Dataflow.Analysis, as: DataflowAnalysis

pipeline = Dataflow.new()
  |> Dataflow.add_source(:sensor, rate: 1000)
  |> Dataflow.add_transform(:parser, latency_ms: 10)
  |> Dataflow.add_sink(:db)
  |> Dataflow.connect(:sensor, :parser)
  |> Dataflow.connect(:parser, :db)

# Highlight high-volume stages
heat_df = DataflowAnalysis.heatmap(pipeline, palette: :heat)

Kino.Layout.grid([
  Kino.VizJS.render(Dataflow.to_dot(heat_df)),
  Kino.VizJS.render(Choreo.to_dot(Analysis.legend(palette: :heat)))
], columns: 2)

5. ThreatModel Risk Hotspots

ThreatModel heatmaps identify security “Hotspots” by node threat density.

alias Choreo.ThreatModel
alias Choreo.ThreatModel.Analysis, as: ThreatAnalysis

model = ThreatModel.new()
  |> ThreatModel.add_trust_boundary("internet", level: 0)
  |> ThreatModel.add_external_entity(:user, boundary: "internet")
  |> ThreatModel.add_process(:api, boundary: "app")
  |> ThreatModel.add_data_store(:db, boundary: "app", sensitivity: :confidential)
  |> ThreatModel.data_flow(:user, :api)
  |> ThreatModel.data_flow(:api, :db)

# Highlight security hotspots
heat_tm = ThreatAnalysis.heatmap(model, palette: :heat)

Kino.Layout.grid([
  Kino.VizJS.render(ThreatModel.to_dot(heat_tm)),
  Kino.VizJS.render(Choreo.to_dot(Analysis.legend(palette: :heat)))
], columns: 2)

6. Transitive Reduction (Dependency Cleanup)

Transitive Reduction simplifies complex dependency graphs by removing redundant edges while preserving reachability. If A -> B, B -> C, and A -> C, the direct edge A -> C is removed because the dependency is already implied by the path through B.

alias Choreo.Dependency

# A dense dependency graph with a redundant edge
deps = Dependency.new()
  |> Dependency.add_application(:api)
  |> Dependency.add_module(:auth)
  |> Dependency.add_library(:phoenix)
  |> Dependency.depends_on(:api, :auth)
  |> Dependency.depends_on(:auth, :phoenix)
  |> Dependency.depends_on(:api, :phoenix) # Redundant!

# Perform reduction
{:ok, reduced} = Analysis.reduce_transitive(deps)

Kino.Layout.grid([
  Kino.VizJS.render(Dependency.to_dot(deps)),
  Kino.VizJS.render(Dependency.to_dot(reduced))
], columns: 2)

7. Path Analysis (Critical Paths & Bottlenecks)

Path Analysis finds the optimal route between two services based on domain-specific metrics.

Fastest Path (Workflow)

Minimize cumulative latency across tasks.

alias Choreo.Workflow

wf = Workflow.new()
  |> Workflow.add_start(:start)
  |> Workflow.add_task(:fast_a, timeout_ms: 10)
  |> Workflow.add_task(:fast_b, timeout_ms: 10)
  |> Workflow.add_task(:slow, timeout_ms: 500)
  |> Workflow.add_end(:done)
  |> Workflow.connect(:start, :fast_a)
  |> Workflow.connect(:fast_a, :fast_b)
  |> Workflow.connect(:fast_b, :done)
  |> Workflow.connect(:start, :slow)
  |> Workflow.connect(:slow, :done)

# Find the "Fastest Path"
{:ok, path} = Analysis.path(wf, :start, :done, measure: :latency)

Kino.VizJS.render(Workflow.to_dot(wf), height: "600px")

Widest Path (Dataflow)

Maximize the bottleneck capacity (minimum throughput) along the path.

alias Choreo.Dataflow

pipeline = Dataflow.new()
  |> Dataflow.add_source(:sensor, rate: 1000)
  |> Dataflow.add_transform(:fast_lane, capacity: 800)
  |> Dataflow.add_transform(:slow_lane, capacity: 50)
  |> Dataflow.add_sink(:db)
  |> Dataflow.connect(:sensor, :fast_lane)
  |> Dataflow.connect(:fast_lane, :db)
  |> Dataflow.connect(:sensor, :slow_lane)
  |> Dataflow.connect(:slow_lane, :db)

# Find the "Widest Path"
{:ok, path} = Analysis.path(pipeline, :sensor, :db, measure: :throughput)

Kino.VizJS.render(Dataflow.to_dot(pipeline, Analysis.highlight(path)))

Custom Weighted Paths

You can also find paths based on arbitrary metadata keys or custom weight functions.

system = Choreo.new()
  |> Choreo.add_service(:a)
  |> Choreo.add_service(:b)
  |> Choreo.add_service(:c)
  |> Choreo.connect(:a, :b, cost: 10)
  |> Choreo.connect(:b, :c, cost: 10)
  |> Choreo.connect(:a, :c, cost: 50)

# Find the cheapest path using the 'cost' variable
{:ok, path} = Analysis.path(system, :a, :c, measure: :cost)

Kino.VizJS.render(Choreo.to_dot(system, Analysis.highlight(path)))