Designing a Quantitative Metric for Microservice Coupling Using Graph Theory and Git History — code snippets

snippet-1.py

import os
import datetime
from collections import defaultdict
from git import Repo
import numpy as np

def analyze_git_temporal_coupling(repo_path, service_dirs, days=90):
    repo = Repo(repo_path)
    since_date = datetime.datetime.now() - datetime.timedelta(days=days)
    
    # Map commit hashes to the set of services modified in that commit
    commit_to_services = defaultdict(set)
    service_commit_counts = defaultdict(int)
    
    # Iterate through commits on the main branch
    for commit in repo.iter_commits('main', since=since_date.isoformat()):
        # Find which services were changed in this commit
        changed_files = list(commit.stats.files.keys())
        for file_path in changed_files:
            for service in service_dirs:
                if file_path.startswith(service + '/'):
                    commit_to_services[commit.hexsha].add(service)
                    
        for service in commit_to_services[commit.hexsha]:
            service_commit_counts[service] += 1
            
    # Calculate Jaccard similarity matrix
    n_services = len(service_dirs)
    jaccard_matrix = np.zeros((n_services, n_services))
    
    for i, s1 in enumerate(service_dirs):
        for j, s2 in enumerate(service_dirs):
            if i == j:
                jaccard_matrix[i][j] = 1.0
                continue
                
            intersection = 0
            union_set = set()
            
            for commit_sha, services in commit_to_services.items():
                if s1 in services and s2 in services:
                    intersection += 1
                if s1 in services or s2 in services:
                    union_set.add(commit_sha)
                    
            union_size = len(union_set)
            jaccard_matrix[i][j] = intersection / union_size if union_size > 0 else 0.0
            
    return jaccard_matrix, service_commit_counts

snippet-2.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"
)

type PromResponse struct {
	Status string `json:"status"`
	Data   struct {
		ResultType string `json:"resultType"`
		Result     []struct {
			Metric map[string]string `json:"metric"`
			Value  []interface{}     `json:"value"`
		} `json:"result"`
	} `json:"data"`
}

type DependencyEdge struct {
	Caller    string  `json:"caller"`
	Callee    string  `json:"callee"`
	CallRate  float64 `json:"call_rate"`
	Transport string  `json:"transport"` // "http", "grpc", "kafka"
}

func FetchRuntimeDependencies(promURL string) ([]DependencyEdge, error) {
	client := &http.Client{Timeout: 10 * time.Second}
	// Query to calculate average request rate between services over the last 6 hours
	query := `sum(rate(calls_total{span_kind="SPAN_KIND_CLIENT"}[6h])) by (service_name, peer_service, transport)`
	url := fmt.Sprintf("%s/api/v1/query?query=%s", promURL, query)

	resp, err := client.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	var promResp PromResponse
	if err := json.NewDecoder(resp.Body).Decode(&promResp); err != nil {
		return nil, err
	}

	var edges []DependencyEdge
	for _, res := range promResp.Data.Result {
		caller := res.Metric["service_name"]
		callee := res.Metric["peer_service"]
		transport := res.Metric["transport"]
		if caller == "" || callee == "" {
			continue
		}
		// Extract value: value is [timestamp, "float_value"]
		if len(res.Value) < 2 {
			continue
		}
		var rate float64
		_, err := fmt.Sscanf(res.Value[1].(string), "%f", &rate)
		if err != nil {
			continue
		}

		edges = append(edges, DependencyEdge{
			Caller:    caller,
			Callee:    callee,
			CallRate:  rate,
			Transport: transport,
		})
	}
	return edges, nil
}

snippet-3.py

import networkx as nx
import numpy as np

def calculate_system_coupling_metrics(services, temporal_matrix, structural_edges, lambda_val=0.6):
    G = nx.DiGraph()
    
    # Add nodes
    for service in services:
        G.add_node(service)
        
    # Map service index
    service_idx = {name: idx for idx, name in enumerate(services)}
    
    # Process structural edges and build Combined Coupling Index
    for edge in structural_edges:
        caller = edge['caller']
        callee = edge['callee']
        if caller not in service_idx or callee not in service_idx:
            continue
            
        # Synchronicity factor
        sigma = 1.0 if edge['transport'] in ['http', 'grpc'] else 0.2
        c_structural = sigma * edge['call_rate']
        
        # Pull temporal coupling Jaccard weight
        i, j = service_idx[caller], service_idx[callee]
        c_temporal = temporal_matrix[i][j]
        
        # Calculate combined coupling weight
        cci = lambda_val * c_temporal + (1 - lambda_val) * c_structural
        
        # Add to directed graph
        G.add_edge(caller, callee, weight=cci)

    # Compute graph metrics
    pagerank = nx.pagerank(G, weight='weight')
    betweenness = nx.betweenness_centrality(G, weight='weight')
    
    # Identify high-risk nodes (hubs) and bottlenecks
    analysis_report = {}
    for node in G.nodes():
        analysis_report[node] = {
            "pagerank_centrality": pagerank[node],
            "betweenness_centrality": betweenness[node],
            "coupling_in_degree": sum([G[u][v]['weight'] for u, v in G.in_edges(node)]),
            "coupling_out_degree": sum([G[u][v]['weight'] for u, v in G.out_edges(node)])
        }
        
    # Calculate Modularity by treating as undirected for community detection
    undirected_G = G.to_undirected()
    try:
        from networkx.algorithms.community import modularity, louvain_communities
        communities = louvain_communities(undirected_G, weight='weight')
        mod_score = modularity(undirected_G, communities, weight='weight')
    except Exception:
        mod_score = 0.0 # Fallback for tiny/disconnected graphs
        
    return analysis_report, mod_score

snippet-4.py

import sys
import json

def evaluate_ci_thresholds(current_metrics_json, baseline_metrics_json, max_allowed_cci=0.45):
    with open(current_metrics_json, 'r') as f:
        current = json.load(f)
    with open(baseline_metrics_json, 'r') as f:
        baseline = json.load(f)
        
    violation_found = False
    violations = []
    
    for service, metrics in current.items():
        base_service = baseline.get(service)
        if not base_service:
            # New service introduced, skip baseline comparison but verify absolute coupling
            if metrics["coupling_out_degree"] > max_allowed_cci:
                violation_found = True
                violations.append(f"New service '{service}' exceeds absolute CCI limit (Out-Degree: {metrics['coupling_out_degree']:.3f} > {max_allowed_cci})")
            continue
            
        # Detect relative degradation in coupling metrics
        delta_in = metrics["coupling_in_degree"] - base_service["coupling_in_degree"]
        delta_out = metrics["coupling_out_degree"] - base_service["coupling_out_degree"]
        
        if delta_out > 0.15:
            violation_found = True
            violations.append(f"Service '{service}' increased its outbound coupling by {delta_out:.3f} (Outbound CCI: {metrics['coupling_out_degree']:.3f})")
            
        if metrics["coupling_out_degree"] > max_allowed_cci:
            violation_found = True
            violations.append(f"Service '{service}' exceeds maximum allowed Outbound CCI ({metrics['coupling_out_degree']:.3f} > {max_allowed_cci})")
            
    if violation_found:
        print("ARCHITECTURE COUPLING GATE FAILED:")
        for v in violations:
            print(f" - {v}")
        sys.exit(1)
        
    print("Architecture coupling validation passed successfully.")
    sys.exit(0)

snippet-5.py

def generate_mermaid_flowchart(services, temporal_matrix, structural_edges, threshold=0.35):
    mermaid_lines = ["graph TD"]
    
    # Class styles for high-risk coupling
    mermaid_lines.append("    classDef risky fill:#ffcccc,stroke:#ff3333,stroke-width:2px;")
    mermaid_lines.append("    classDef normal fill:#e1f5fe,stroke:#039be5,stroke-width:1px;")
    
    service_idx = {name: idx for idx, name in enumerate(services)}
    risky_services = set()
    
    # Render edges
    for edge in structural_edges:
        caller = edge['caller']
        callee = edge['callee']
        
        i, j = service_idx[caller], service_idx[callee]
        temporal_val = temporal_matrix[i][j]
        
        # Calculate approximate coupling weight
        cci = 0.6 * temporal_val + 0.4 * edge['call_rate']
        
        if cci > threshold:
            risky_services.add(caller)
            risky_services.add(callee)
            edge_style = f" ===|CCI: {cci:.2f}| "
        else:
            edge_style = f" --->|CCI: {cci:.2f}| "
            
        mermaid_lines.append(f"    {caller}{edge_style}{callee}")
        
    # Apply class styles
    for service in services:
        if service in risky_services:
            mermaid_lines.append(f"    class {service} risky;")
        else:
            mermaid_lines.append(f"    class {service} normal;")
            
    return "\n".join(mermaid_lines)

snippet-6.yaml

name: Architectural Coupling Verification

on:
  pull_request:
    branches:
      - main

jobs:
  analyze-coupling:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout Code
        uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'

      - name: Install Dependencies
        run: |
          pip install GitPython numpy networkx requests

      - name: Fetch Baseline Metrics
        run: |
          git show origin/main:ci/coupling_baseline.json > baseline.json || echo "{}" > baseline.json

      - name: Calculate Current Coupling
        env:
          PROMETHEUS_URL: "https://prometheus.production.internal"
        run: |
          python ci/calculate_coupling.py --output current.json

      - name: Evaluate Architecture Gates
        run: |
          python ci/evaluate_thresholds.py --current current.json --baseline baseline.json --max-cci 0.45

      - name: Generate Visual Graph
        if: always()
        run: |
          python ci/generate_mermaid.py --output mermaid.md

      - name: Post PR Comment
        uses: mshick/bootstrap-github-actions/post-comment@v1
        if: always()
        with:
          github-token: ${{ secrets.GITHUB_TOKEN }}
          path: mermaid.md