Service Dependency Mapping and Failure Domain Analysis: From Topology Discovery to Blast Radius Control

Overview

In modern microservice architectures, a seemingly simple user request may traverse dozens of service nodes. When an incident occurs, the first question an SRE engineer faces is often not “how to fix it” but “what is the scope of impact.” Without a fast answer to this question, incident recovery gets bogged down in endless investigation.

Service Dependency Maps and Failure Domain Analysis are the engineering methodologies that address this problem. The former solves the cognitive problem of “who depends on whom and how,” while the latter tackles the control problem of “how far will a failure spread and how large is the blast radius.” Together, they form the foundational infrastructure of SRE reliability engineering.

This article starts with methods for discovering dependency topologies, dives deep into failure domain identification and isolation strategies, and concludes with engineering practices for blast radius control.

The Complexity of Service Dependencies

Dependency Characteristics in Microservice Architectures

In the monolithic era, dependency relationships were explicit and compile-time — you could fully depict the dependency graph through import statements and function calls. Microservice architectures have fundamentally changed this paradigm:

Dimension	Monolithic Application	Microservice Architecture
Discovery method	Static code analysis	Runtime traffic observation
Dependency type	Function calls	HTTP/gRPC/Message queues/Event buses
Stability	Determined at compile time	Dynamically changes at runtime
Visibility	IDE can navigate directly	Requires specialized tooling
Failure propagation	In-process exception stacks	Cross-network cascading failures
Dependency scale	Tens to hundreds	Hundreds to thousands

Dependency Classification System

Not all dependencies carry the same risk level. A mature dependency map must annotate dependencies with classification labels:

By invocation method:

Synchronous calls: HTTP REST, gRPC, database queries. The caller blocks waiting for a response, making these the primary propagation path for cascading failures.
Asynchronous calls: Message queues (Kafka, RabbitMQ), event buses. The caller does not block, but consumer failures can lead to message accumulation.
Shared resource dependencies: Shared databases, cache clusters, storage volumes. Resource contention can cause indirect failures.
Infrastructure dependencies: DNS, service discovery, configuration centers. Failures in these affect a very wide scope and are on the critical path.

By criticality:

Strong dependency: When the callee is unavailable, the caller cannot complete its core function. For example, the order service depends on the inventory service.
Weak dependency: When the callee is unavailable, the caller can degrade gracefully. For example, the product detail page depends on the recommendation service.
Conditional dependency: A dependency that only triggers under specific scenarios. For example, the coupon service is only called during promotional campaigns.

# Dependency classification example
class DependencyType:
    SYNC_HTTP = "sync_http"
    SYNC_GRPC = "sync_grpc"
    ASYNC_MQ = "async_mq"
    SHARED_DB = "shared_db"
    SHARED_CACHE = "shared_cache"
    INFRA_DNS = "infra_dns"
    INFRA_SERVICE_DISCOVERY = "infra_sd"

class DependencyCriticality:
    STRONG = "strong"       # Non-degradable
    WEAK = "weak"           # Degradeable
    CONDITIONAL = "conditional"  # Condition-triggered

# Dependency relationship data structure
class ServiceDependency:
    def __init__(self, caller, callee, dep_type, criticality):
        self.caller = caller               # Calling service name
        self.callee = callee               # Called service name
        self.dep_type = dep_type           # Dependency type
        self.criticality = criticality     # Criticality level
        self.slo_latency_ms = None         # Dependency call P99 latency
        self.error_rate = None             # Dependency call error rate
        self.fallback_enabled = False       # Whether fallback is configured
        self.circuit_breaker = False       # Whether circuit breaker is configured

Dependency Topology Discovery Methods

Static Discovery: Extracting from Code and Configuration

Static discovery builds the dependency graph by analyzing code repositories and deployment configurations. Its advantage is complete coverage (including rarely invoked paths), while its disadvantage is the inability to reflect actual runtime traffic.

Extracting from Kubernetes configurations:

# Discovering dependencies through Service and Endpoint relationships
# order-service's Deployment references inventory-service
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: order-service
        env:
        - name: INVENTORY_SERVICE_URL
          value: "http://inventory-service.production.svc.cluster.local:8080"
        - name: PAYMENT_SERVICE_URL
          value: "http://payment-service.production.svc.cluster.local:8090"
        - name: KAFKA_BROKERS
          value: "kafka-broker.data.svc.cluster.local:9092"

#!/usr/bin/env python3
"""Extract service dependencies from Kubernetes ConfigMaps and Deployments"""

import yaml
import re
import json
from collections import defaultdict

class K8sDependencyExtractor:
    """Extract inter-service dependencies from K8s configurations"""

    # Regex matching K8s internal service DNS
    SERVICE_DNS_PATTERN = re.compile(
        r'(?:https?://)?([a-z0-9-]+)\.([a-z0-9-]+)\.svc\.cluster\.local(?::(\d+))?'
    )
    # Regex matching service references in environment variables
    ENV_SERVICE_PATTERN = re.compile(
        r'(?:https?://)?([a-z0-9-]+):(\d+)'
    )

    def __init__(self):
        self.dependencies = defaultdict(list)

    def extract_from_manifest(self, manifest_text):
        """Extract dependencies from YAML manifest text"""
        docs = list(yaml.safe_load_all(manifest_text))

        for doc in docs:
            if not doc or doc.get('kind') not in ('Deployment', 'ConfigMap'):
                continue

            name = doc.get('metadata', {}).get('name', '')
            namespace = doc.get('metadata', {}).get('namespace', 'default')

            if doc['kind'] == 'Deployment':
                self._extract_from_deployment(name, namespace, doc)
            elif doc['kind'] == 'ConfigMap':
                self._extract_from_configmap(name, namespace, doc)

        return dict(self.dependencies)

    def _extract_from_deployment(self, name, namespace, doc):
        """Extract service references from Deployment environment variables"""
        containers = (
            doc.get('spec', {})
            .get('template', {})
            .get('spec', {})
            .get('containers', [])
        )

        for container in containers:
            env_vars = container.get('env', [])
            for env in env_vars:
                value = str(env.get('value', ''))
                # Find svc.cluster.local format service references
                matches = self.SERVICE_DNS_PATTERN.findall(value)
                for svc_name, svc_ns, port in matches:
                    self.dependencies[name].append({
                        'callee': svc_name,
                        'namespace': svc_ns or namespace,
                        'port': port or '80',
                        'source': 'env_var',
                        'env_key': env.get('name', '')
                    })

    def _extract_from_configmap(self, name, namespace, doc):
        """Extract service references from ConfigMap data"""
        data = doc.get('data', {})
        for key, value in data.items():
            if not isinstance(value, str):
                continue
            matches = self.SERVICE_DNS_PATTERN.findall(value)
            for svc_name, svc_ns, port in matches:
                self.dependencies[name].append({
                    'callee': svc_name,
                    'namespace': svc_ns or namespace,
                    'port': port or '80',
                    'source': 'configmap',
                    'config_key': key
                })

    def to_graph(self):
        """Output JSON representation of the dependency graph"""
        nodes = set()
        edges = []
        for caller, deps in self.dependencies.items():
            nodes.add(caller)
            for dep in deps:
                nodes.add(dep['callee'])
                edges.append({
                    'source': caller,
                    'target': dep['callee'],
                    'type': dep.get('source', 'unknown'),
                    'port': dep.get('port', '')
                })
        return {
            'nodes': sorted(list(nodes)),
            'edges': edges,
            'total_services': len(nodes),
            'total_dependencies': len(edges)
        }

# Usage example
if __name__ == '__main__':
    sample_manifest = """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: order-service
        env:
        - name: INVENTORY_SERVICE_URL
          value: "http://inventory-service.production.svc.cluster.local:8080"
        - name: PAYMENT_SERVICE_URL
          value: "http://payment-service.production.svc.cluster.local:8090"
"""
    extractor = K8sDependencyExtractor()
    extractor.extract_from_manifest(sample_manifest)
    print(json.dumps(extractor.to_graph(), indent=2, ensure_ascii=False))

Dynamic Discovery: Runtime Traffic Observation

Dynamic discovery builds the dependency graph by observing actual runtime traffic, reflecting real invocation relationships. There are three mainstream approaches:

Method	Principle	Advantage	Disadvantage
Distributed tracing	Span parent-child relationships in traces	Request-level precision, includes latency data	Requires SDK integration, sampling rate limits
Service Mesh	Sidecar proxies intercept traffic	Non-intrusive, covers all L7 traffic	Limited to mesh-managed services
eBPF	Kernel-level network call interception	Non-intrusive, covers all network traffic	High technical barrier, requires newer kernel

Trace analysis based on Jaeger/OpenTelemetry:

#!/usr/bin/env python3
"""Extract service dependencies from OpenTelemetry / Jaeger Trace data"""

import json
from collections import defaultdict
from datetime import datetime, timedelta

class TraceDependencyExtractor:
    """Extract service dependency graph from distributed tracing data"""

    def __init__(self):
        # Dependencies: {(caller, callee): {count, p99_latency, error_count}}
        self.dependencies = defaultdict(lambda: {
            'call_count': 0,
            'latencies': [],
            'error_count': 0,
            'last_seen': None
        })

    def process_trace(self, trace_data):
        """Process a single Trace to extract span parent-child relationships"""
        spans = trace_data.get('spans', [])

        # Build span_id -> span mapping
        span_map = {s['spanID']: s for s in spans}

        for span in spans:
            parent_id = span.get('parentSpanID')
            if not parent_id or parent_id not in span_map:
                continue

            parent = span_map[parent_id]

            # Extract service name (from process/tag info)
            caller_service = self._get_service_name(parent)
            callee_service = self._get_service_name(span)

            if not caller_service or not callee_service:
                continue
            if caller_service == callee_service:
                continue  # Skip intra-service calls

            key = (caller_service, callee_service)
            dep = self.dependencies[key]

            dep['call_count'] += 1

            # Record latency
            duration_us = span.get('duration', 0)
            dep['latencies'].append(duration_us)

            # Record errors
            tags = span.get('tags', [])
            for tag in tags:
                if (tag.get('key') == 'error' and
                    tag.get('value') is True):
                    dep['error_count'] += 1
                    break

            # Update last seen time
            start_time = span.get('startTime', 0)
            if start_time:
                dep['last_seen'] = start_time

    def _get_service_name(self, span):
        """Extract service name from span's process information"""
        process_id = span.get('processID')
        processes = span.get('processes', {})
        process = processes.get(process_id, {})
        tags = process.get('tags', [])

        for tag in tags:
            if tag.get('key') == 'service.name':
                return tag.get('value')
        return process.get('serviceName')

    def build_dependency_graph(self):
        """Build the final service dependency graph"""
        edges = []
        for (caller, callee), data in self.dependencies.items():
            latencies = sorted(data['latencies'])
            p99_index = int(len(latencies) * 0.99) if latencies else 0
            p99_latency = latencies[p99_index] if latencies else 0

            error_rate = (
                data['error_count'] / data['call_count']
                if data['call_count'] > 0 else 0
            )

            edges.append({
                'source': caller,
                'target': callee,
                'call_count': data['call_count'],
                'p99_latency_ms': round(p99_latency / 1000, 2),
                'error_rate': round(error_rate, 4),
                'last_seen': data['last_seen']
            })

        # Sort by call count
        edges.sort(key=lambda x: x['call_count'], reverse=True)

        nodes = set()
        for e in edges:
            nodes.add(e['source'])
            nodes.add(e['target'])

        return {
            'nodes': sorted(list(nodes)),
            'edges': edges,
            'total_services': len(nodes),
            'total_edges': len(edges),
            'generated_at': datetime.utcnow().isoformat()
        }

# Usage example
if __name__ == '__main__':
    # Simulate a Trace
    sample_trace = {
        'traceID': 'abc123',
        'spans': [
            {
                'spanID': 'span1',
                'parentSpanID': None,
                'operationName': 'GET /api/orders',
                'startTime': 1752216000000000,
                'duration': 50000,
                'processID': 'p1',
                'tags': []
            },
            {
                'spanID': 'span2',
                'parentSpanID': 'span1',
                'operationName': 'GET /api/inventory',
                'startTime': 1752216000100000,
                'duration': 12000,
                'processID': 'p2',
                'tags': []
            },
            {
                'spanID': 'span3',
                'parentSpanID': 'span1',
                'operationName': 'POST /api/payment',
                'startTime': 1752216000200000,
                'duration': 30000,
                'processID': 'p3',
                'tags': [{'key': 'error', 'value': True}]
            }
        ],
        'processes': {
            'p1': {'serviceName': 'order-service', 'tags': []},
            'p2': {'serviceName': 'inventory-service', 'tags': []},
            'p3': {'serviceName': 'payment-service', 'tags': []}
        }
    }

    extractor = TraceDependencyExtractor()
    extractor.process_trace(sample_trace)
    graph = extractor.build_dependency_graph()
    print(json.dumps(graph, indent=2, ensure_ascii=False))

Non-intrusive topology discovery with eBPF:

The eBPF approach does not require application code changes. It intercepts network calls at the kernel level, making it suitable as a comprehensive fallback for dependency discovery:

# Use bpftrace to capture TCP connection relationships
# This script outputs all newly established TCP connections' source process and target address
#!/usr/bin/env bpftrace

BEGIN
{
    printf("Tracing TCP connections... Ctrl-C to stop.\n");
    printf("%-12s %-16s %-6s %-16s %-6s\n",
           "TIME", "COMM", "PID", "DADDR", "DPORT");
}

kprobe:tcp_connect
{
    $sk = (struct sock *)arg0;
    $daddr = $sk->sk_daddr;

    time("%H:%M:%S   ");
    printf("%-16s %-6d ", comm, pid);
    printf("%-16s %-6d\n",
           ntop($daddr),
           $sk->sk_dport >> 8);
}

# Discover inter-Pod communication using kubectl + eBPF toolchain
# Service dependency map based on Cilium Hubble
hubble observe --follow \
  --type l3/4 \
  --output json | jq '{
    source: .source.podName,
    destination: .destination.podName,
    port: .destination.port,
    protocol: .l4.protocol,
    verdict: .verdict
  }' | jq -s 'group_by(.source + "->" + .destination) | map({
    edge: .[0].source + " -> " + .[0].destination,
    count: length,
    ports: [.[].port] | unique
  })'

Complementary Strategy: Static and Dynamic Discovery

Both methods have limitations and should be combined in production:

Discovery Dimension	Static Discovery	Dynamic Discovery	Complementary Value
Coverage completeness	High (includes rare paths)	Limited by sampling rate	Static fills dynamic gaps
Dependency accuracy	Low (includes deprecated configs)	High (actual calls)	Dynamic filters static noise
Real-time capability	None	Seconds-level	Dynamic senses architecture changes
Resource overhead	Minimal	Medium to high	Static serves as baseline
Operational threshold	Low	High	Choose based on need

class HybridDependencyGraph:
    """Hybrid dependency graph merging static and dynamic discovery results"""

    def __init__(self):
        self.static_edges = {}   # Statically discovered edges
        self.dynamic_edges = {}  # Dynamically discovered edges
        self.merged_graph = {}   # Merged graph

    def add_static_dependency(self, caller, callee, source='config'):
        """Add a statically discovered dependency"""
        key = (caller, callee)
        if key not in self.static_edges:
            self.static_edges[key] = {
                'source': source,
                'verified': False
            }

    def add_dynamic_dependency(self, caller, callee, call_count,
                               p99_latency_ms, error_rate):
        """Add a dynamically discovered dependency"""
        key = (caller, callee)
        self.dynamic_edges[key] = {
            'call_count': call_count,
            'p99_latency_ms': p99_latency_ms,
            'error_rate': error_rate,
            'verified': True
        }

    def merge(self):
        """Merge static and dynamic discovery results"""
        all_keys = set(self.static_edges.keys()) | set(self.dynamic_edges.keys())

        for key in all_keys:
            caller, callee = key
            static = self.static_edges.get(key, {})
            dynamic = self.dynamic_edges.get(key, {})

            edge = {
                'caller': caller,
                'callee': callee,
                'static_found': key in self.static_edges,
                'dynamic_found': key in self.dynamic_edges,
                'call_count': dynamic.get('call_count', 0),
                'p99_latency_ms': dynamic.get('p99_latency_ms', None),
                'error_rate': dynamic.get('error_rate', None),
                'status': self._classify_edge(static, dynamic),
                'source': static.get('source', 'runtime')
            }
            self.merged_graph[key] = edge

        return self.merged_graph

    def _classify_edge(self, static, dynamic):
        """Classify and label edges"""
        if static and dynamic:
            return 'verified'         # Static config and runtime confirmed
        elif not static and dynamic:
            return 'undocumented'      # Exists at runtime but not in config
        elif static and not dynamic:
            return 'dormant'           # In config but not invoked at runtime
        return 'unknown'

    def get_risk_edges(self):
        """Get edges that need attention"""
        risks = []
        for key, edge in self.merged_graph.items():
            if edge['status'] == 'undocumented':
                risks.append({
                    'edge': f"{edge['caller']} -> {edge['callee']}",
                    'risk': 'Undocumented dependency, may be missed during architecture changes',
                    'severity': 'medium'
                })
            elif (edge['status'] == 'verified' and
                  edge['error_rate'] and edge['error_rate'] > 0.05):
                risks.append({
                    'edge': f"{edge['caller']} -> {edge['callee']}",
                    'risk': f"Error rate {edge['error_rate']:.1%}, needs investigation",
                    'severity': 'high'
                })
        return risks

Failure Domain Analysis

What Is a Failure Domain

A failure domain is the set of components and services affected when a single component fails. Understanding failure domains is fundamentally about understanding failure propagation paths.

“Failures do not stay at their origin. A database connection pool exhaustion can cause dozens of upstream services to cascade into timeouts. A DNS misconfiguration can paralyze an entire datacenter. Controlling failure domain boundaries is controlling the total amount of system risk.” — Reference: Google SRE Book, Chapter 6

Hierarchical Model of Failure Domains

Failure domains are nested in layers, with each outer layer encompassing a wider impact scope:

┌─────────────────────────────────────────────────┐
│  Global Failure Domain                            │
│  ┌───────────────────────────────────────────┐  │
│  │  Region Failure Domain                      │  │
│  │  ┌─────────────────────────────────────┐  │  │
│  │  │  Availability Zone (AZ) Domain        │  │  │
│  │  │  ┌───────────────────────────────┐   │  │  │
│  │  │  │  Cluster Failure Domain        │   │  │  │
│  │  │  │  ┌─────────────────────────┐  │   │  │  │
│  │  │  │  │  Node Failure Domain     │  │   │  │  │
│  │  │  │  │  ┌───────────────────┐  │  │   │  │  │
│  │  │  │  │  │ Pod Failure Domain │  │  │   │  │  │
│  │  │  │  │  │  ┌─────────────┐  │  │  │   │  │  │
│  │  │  │  │  │  │ Container    │  │  │  │   │  │  │
│  │  │  │  │  │  └─────────────┘  │  │  │   │  │  │
│  │  │  │  │  └───────────────────┘  │  │   │  │  │
│  │  │  │  └─────────────────────────┘  │   │  │  │
│  │  │  └───────────────────────────────┘  │  │  │
│  │  └─────────────────────────────────────┘ │  │
│  └───────────────────────────────────────────┘  │
└─────────────────────────────────────────────────┘

Level	Typical Failure Cause	Impact Scope	Isolation Method
Container	OOM, application exception	Single container	Restart policy, health checks
Pod	Node eviction, scheduling failure	Single Pod replica	Multiple replicas, PDB
Node	Hardware failure, kernel panic	All Pods on the node	Node isolation, anti-affinity
Cluster	Control plane failure, network partition	All services in cluster	Multi-cluster, federation
AZ	Power outage, network interruption	All resources in AZ	Multi-AZ deployment, cross-AZ load balancing
Region	Regional failure	All resources in region	Multi-region active-active
Global	DNS failure, certificate expiry	Entire site	Disaster recovery switchover, degradation playbook

Failure Propagation Path Analysis

Failure propagation follows the edges of the dependency graph. Analyzing propagation paths requires answering three questions:

Where does the failure start: Identify the location of the root cause service
Who will be affected: Perform reachability analysis along dependency graph edges
How severe is the impact: Assess impact severity based on dependency type and criticality

#!/usr/bin/env python3
"""Failure domain analysis engine: compute failure propagation paths and blast radius"""

from collections import deque, defaultdict
import json

class FailureDomainAnalyzer:
    """Analyze failure propagation and blast radius based on service dependency graph"""

    def __init__(self, dependency_graph):
        """
        dependency_graph: {
            'edges': [
                {'source': 'A', 'target': 'B', 'criticality': 'strong', ...},
                ...
            ]
        }
        """
        self.graph = defaultdict(list)
        self.reverse_graph = defaultdict(list)
        self.edge_info = {}

        for edge in dependency_graph.get('edges', []):
            src, dst = edge['source'], edge['target']
            self.graph[src].append(dst)
            self.reverse_graph[dst].append(src)

            key = (src, dst)
            self.edge_info[key] = {
                'criticality': edge.get('criticality', 'strong'),
                'fallback': edge.get('fallback_enabled', False),
                'circuit_breaker': edge.get('circuit_breaker', False),
                'call_count': edge.get('call_count', 0)
            }

    def analyze_blast_radius(self, failed_service, max_depth=10):
        """
        Analyze the blast radius of a single service failure

        Args:
            failed_service: Name of the failed service
            max_depth: Maximum propagation depth

        Returns:
            {
                'affected_services': [...],     # List of affected services
                'propagation_paths': [...],     # Propagation paths
                'blast_radius_score': float,    # Blast radius score (0-100)
                'critical_path': bool           # Whether critical path is affected
            }
        """
        affected = set()
        propagation_paths = []
        visited = set()

        # BFS traversal of the reverse dependency graph (who depends on the failed service)
        queue = deque()
        queue.append((failed_service, 0, []))

        while queue:
            service, depth, path = queue.popleft()

            if depth > max_depth:
                continue

            if service in visited:
                continue
            visited.add(service)

            current_path = path + [service]

            if service != failed_service:
                affected.add(service)
                if len(current_path) > 1:
                    propagation_paths.append({
                        'path': ' -> '.join(current_path),
                        'depth': depth,
                        'edge_info': self._get_path_info(current_path)
                    })

            # Traverse upstream along the reverse dependency graph
            for caller in self.reverse_graph.get(service, []):
                edge_key = (caller, service)
                edge_data = self.edge_info.get(edge_key, {})

                # If fallback or circuit breaker exists, propagation is truncated here
                if (edge_data.get('fallback') or
                    edge_data.get('circuit_breaker')):
                    # Record truncation point
                    propagation_paths.append({
                        'path': ' -> '.join(current_path + [caller]),
                        'depth': depth + 1,
                        'truncated': True,
                        'truncation_reason': (
                            'fallback' if edge_data.get('fallback')
                            else 'circuit_breaker'
                        )
                    })
                    continue

                queue.append((caller, depth + 1, current_path))

        # Calculate blast radius score
        blast_radius = self._calculate_blast_radius(
            failed_service, affected
        )

        # Determine whether critical path is affected
        critical = self._is_critical_path(failed_service, affected)

        return {
            'failed_service': failed_service,
            'affected_services': sorted(list(affected)),
            'affected_count': len(affected),
            'propagation_paths': propagation_paths,
            'blast_radius_score': blast_radius,
            'critical_path': critical
        }

    def _get_path_info(self, path):
        """Get info for each edge along the propagation path"""
        info = []
        for i in range(len(path) - 1):
            key = (path[i], path[i + 1])
            edge = self.edge_info.get(key, {})
            info.append({
                'from': path[i],
                'to': path[i + 1],
                'criticality': edge.get('criticality', 'unknown'),
                'fallback': edge.get('fallback', False),
                'circuit_breaker': edge.get('circuit_breaker', False)
            })
        return info

    def _calculate_blast_radius(self, failed_service, affected_services):
        """Calculate blast radius score (0-100)"""
        if not affected_services:
            return 0

        score = 0
        for svc in affected_services:
            # Each affected service contributes base score
            score += 5
            # If the service is depended on by many others, increase weight
            dependents = len(self.reverse_graph.get(svc, []))
            score += dependents * 2

        # Clamp to 0-100 range
        return min(score, 100)

    def _is_critical_path(self, failed_service, affected_services):
        """Determine whether the critical business path is affected"""
        critical_services = {'api-gateway', 'order-service',
                              'payment-service', 'auth-service'}
        all_affected = affected_services | {failed_service}
        return bool(all_affected & critical_services)

    def find_single_points_of_failure(self):
        """Identify single points of failure"""
        spof = []
        total_services = set(self.graph.keys()) | set(self.reverse_graph.keys())

        for service in total_services:
            dependents = self.reverse_graph.get(service, [])
            if len(dependents) == 0:
                continue  # Not depended on, not a single point

            # Check whether degradation protection exists
            all_protected = True
            for caller in dependents:
                edge_key = (caller, service)
                edge_data = self.edge_info.get(edge_key, {})
                if not (edge_data.get('fallback') or
                       edge_data.get('circuit_breaker')):
                    all_protected = False
                    break

            if not all_protected:
                spof.append({
                    'service': service,
                    'dependent_count': len(dependents),
                    'dependents': list(dependents),
                    'risk': 'high' if len(dependents) > 5 else 'medium'
                })

        spof.sort(key=lambda x: x['dependent_count'], reverse=True)
        return spof

# Usage example
if __name__ == '__main__':
    # Simulate dependency graph
    dep_graph = {
        'edges': [
            {'source': 'api-gateway', 'target': 'order-service',
             'criticality': 'strong'},
            {'source': 'order-service', 'target': 'inventory-service',
             'criticality': 'strong'},
            {'source': 'order-service', 'target': 'payment-service',
             'criticality': 'strong'},
            {'source': 'order-service', 'target': 'recommendation-service',
             'criticality': 'weak', 'fallback_enabled': True},
            {'source': 'payment-service', 'target': 'fraud-detection',
             'criticality': 'strong'},
            {'source': 'payment-service', 'target': 'notification-service',
             'criticality': 'weak', 'fallback_enabled': True},
            {'source': 'inventory-service', 'target': 'product-service',
             'criticality': 'strong'},
            {'source': 'product-service', 'target': 'cache-cluster',
             'criticality': 'strong'},
        ]
    }

    analyzer = FailureDomainAnalyzer(dep_graph)

    # Analyze impact of inventory-service failure
    result = analyzer.analyze_blast_radius('inventory-service')
    print(json.dumps(result, indent=2, ensure_ascii=False))

    print("\n--- Single Point of Failure Analysis ---")
    spof = analyzer.find_single_points_of_failure()
    print(json.dumps(spof, indent=2, ensure_ascii=False))

Blast Radius Control Strategies

The core idea of controlling blast radius is to install “firewalls” along dependency paths so that failure propagation is truncated as early as possible.

Strategy 1: Circuit Breaker Pattern

# Circuit breaker configuration in Istio DestinationRule
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service-cb
  namespace: production
spec:
  host: order-service.production.svc.cluster.local
  trafficPolicy:
    outlierDetection:
      # Trip after 5 consecutive 5xx errors
      consecutive5xxErrors: 5
      # Detection interval 30 seconds
      interval: 30s
      # Base ejection time 30 seconds
      baseEjectionTime: 30s
      # Maximum ejection ratio 50%
      maxEjectionPercent: 50
      # Minimum healthy instance percentage
      minHealthPercent: 50
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 10
        maxRetries: 2
        # Idle timeout
        idleTimeout: 60s

Strategy 2: Degradation and Fallback

#!/usr/bin/env python3
"""Service call degradation framework"""

import time
import logging
from functools import wraps
from typing import Any, Callable, Optional

logger = logging.getLogger(__name__)

class CircuitBreaker:
    """Simple circuit breaker implementation"""

    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
        self.state = 'closed'  # closed, open, half_open

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if self.state == 'open':
                if self._should_try_reset():
                    self.state = 'half_open'
                    logger.info(f"Circuit breaker half-open for {func.__name__}")
                else:
                    raise CircuitBreakerOpenError(
                        f"Circuit breaker is open for {func.__name__}"
                    )

            try:
                result = func(*args, **kwargs)
                self._on_success()
                return result
            except Exception as e:
                self._on_failure()
                raise

        return wrapper

    def _should_try_reset(self):
        if self.last_failure_time is None:
            return True
        return time.time() - self.last_failure_time > self.recovery_timeout

    def _on_success(self):
        self.failure_count = 0
        if self.state == 'half_open':
            self.state = 'closed'
            logger.info("Circuit breaker closed")

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if (self.failure_count >= self.failure_threshold and
            self.state != 'open'):
            self.state = 'open'
            logger.warning(
                f"Circuit breaker opened after "
                f"{self.failure_count} failures"
            )

class CircuitBreakerOpenError(Exception):
    """Circuit breaker open exception"""
    pass

def with_fallback(fallback_func: Optional[Callable] = None,
                  default_value: Any = None):
    """
    Fallback decorator: return default value or execute fallback
    when the primary call fails

    Args:
        fallback_func: Function to execute during fallback
        default_value: Default return value when no fallback is available
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except CircuitBreakerOpenError:
                logger.warning(
                    f"Circuit breaker open for {func.__name__}, "
                    f"using fallback"
                )
                if fallback_func:
                    return fallback_func(*args, **kwargs)
                return default_value
            except Exception as e:
                logger.error(
                    f"{func.__name__} failed: {e}, using fallback"
                )
                if fallback_func:
                    return fallback_func(*args, **kwargs)
                return default_value

        return wrapper
    return decorator

# Real-world usage example
class OrderService:
    """Order service demonstrating combined degradation strategies"""

    def __init__(self):
        self.inventory_cb = CircuitBreaker(
            failure_threshold=5, recovery_timeout=30
        )
        self.recommendation_cb = CircuitBreaker(
            failure_threshold=3, recovery_timeout=60
        )

    @CircuitBreaker(failure_threshold=5, recovery_timeout=30)
    def call_inventory(self, product_id):
        """Call inventory service (strong dependency, no fallback)"""
        # Simulate call
        raise ConnectionError("inventory-service unavailable")

    @with_fallback(
        default_value={"product_id": None, "quantity": 0,
                       "available": False}
    )
    def call_recommendation(self, user_id):
        """Call recommendation service (weak dependency, fallback to empty)"""
        # Simulate call
        raise ConnectionError("recommendation-service unavailable")

    def place_order(self, user_id, product_id, quantity):
        """Order placement flow"""
        # Strong dependency: inventory check failure fails the entire flow
        try:
            inventory = self.call_inventory(product_id)
            if inventory['quantity'] < quantity:
                return {'success': False, 'reason': 'insufficient_stock'}
        except CircuitBreakerOpenError:
            return {
                'success': False,
                'reason': 'inventory_service_unavailable',
                'retry_after': 30
            }

        # Weak dependency: recommendation failure does not block ordering
        recommendations = self.call_recommendation(user_id)

        return {
            'success': True,
            'order_id': f"ORD-{time.time()}",
            'recommendations': recommendations
        }

Strategy 3: Bulkhead Isolation

# Resource isolation through ResourceQuota and LimitRange in Kubernetes
# Ensures that resource exhaustion in one namespace does not affect others

---
apiVersion: v1
kind: ResourceQuota
metadata:
  name: team-payments-quota
  namespace: payments
spec:
  hard:
    requests.cpu: "20"
    requests.memory: 40Gi
    limits.cpu: "40"
    limits.memory: 80Gi
    pods: "50"
    services: "10"
    configmaps: "20"

---
apiVersion: v1
kind: LimitRange
metadata:
  name: payments-limits
  namespace: payments
spec:
  limits:
  # Resource upper and lower bounds per Pod
  - type: Container
    default:
      cpu: "500m"
      memory: "512Mi"
    defaultRequest:
      cpu: "100m"
      memory: "128Mi"
    max:
      cpu: "4"
      memory: "8Gi"
    min:
      cpu: "50m"
      memory: "64Mi"
  # Size limit per PVC
  - type: PersistentVolumeClaim
    max:
      storage: 100Gi
    min:
      storage: 1Gi

#!/usr/bin/env python3
"""Thread pool isolation implementing the bulkhead pattern"""

import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from collections import defaultdict

class BulkheadManager:
    """
    Bulkhead pattern: allocate independent thread pools for different service calls
    Exhaustion of one service's thread pool does not affect other services
    """

    def __init__(self):
        self.executors = {}
        self.semaphores = {}
        self.metrics = defaultdict(lambda: {
            'total_calls': 0,
            'rejected_calls': 0,
            'active_calls': 0
        })
        self._lock = threading.Lock()

    def register_service(self, service_name, max_concurrent=20):
        """Register an independent thread pool for a service"""
        self.executors[service_name] = ThreadPoolExecutor(
            max_workers=max_concurrent,
            thread_name_prefix=f"bulkhead-{service_name}"
        )
        # Use semaphore for concurrency control
        import threading as th
        self.semaphores[service_name] = th.Semaphore(max_concurrent)

    def call(self, service_name, func, *args, **kwargs):
        """Call a service through a bulkhead-isolated thread pool"""
        if service_name not in self.executors:
            raise ValueError(f"Service {service_name} not registered")

        sem = self.semaphores[service_name]

        # Try to acquire semaphore (non-blocking)
        acquired = sem.acquire(blocking=False)
        if not acquired:
            with self._lock:
                self.metrics[service_name]['rejected_calls'] += 1
            raise BulkheadFullError(
                f"Bulkhead for {service_name} is full, "
                f"max_concurrent reached"
            )

        try:
            with self._lock:
                self.metrics[service_name]['active_calls'] += 1
                self.metrics[service_name]['total_calls'] += 1

            executor = self.executors[service_name]
            future = executor.submit(func, *args, **kwargs)
            return future.result()
        finally:
            sem.release()
            with self._lock:
                self.metrics[service_name]['active_calls'] -= 1

    def get_metrics(self):
        """Get bulkhead status for each service"""
        with self._lock:
            return dict(self.metrics)

class BulkheadFullError(Exception):
    """Bulkhead full exception"""
    pass

# Usage example
if __name__ == '__main__':
    manager = BulkheadManager()

    # Allocate different concurrency limits for different services
    manager.register_service('payment', max_concurrent=10)
    manager.register_service('inventory', max_concurrent=20)
    manager.register_service('recommendation', max_concurrent=5)

    def simulate_call(service, duration=0.1):
        time.sleep(duration)
        return f"{service} call succeeded"

    # Simulate concurrent calls
    threads = []
    results = []
    errors = []

    def worker(service):
        try:
            result = manager.call(service, simulate_call, service, 0.5)
            results.append(result)
        except BulkheadFullError as e:
            errors.append(str(e))

    # High volume of concurrent calls to payment service
    for i in range(15):
        t = threading.Thread(target=worker, args=('payment',))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    print(f"Successes: {len(results)}")
    print(f"Rejections: {len(errors)}")
    print(f"Metrics: {manager.get_metrics()}")

Continuous Maintenance of Dependency Maps

Automated Topology Discovery Pipeline

A dependency map is not a one-time artifact. It needs continuous updates to reflect architecture changes:

#!/usr/bin/env python3
"""
Dependency map continuous update pipeline
Periodically collect dependency data from multiple sources,
merge and compare with historical versions to detect architecture changes
"""

import json
import os
from datetime import datetime, timedelta
from collections import defaultdict

class DependencyMapPipeline:
    """Dependency map update pipeline"""

    def __init__(self, storage_path='.dumate/dependency-maps'):
        self.storage_path = storage_path
        os.makedirs(storage_path, exist_ok=True)

    def run(self, static_deps, dynamic_deps):
        """Execute the full pipeline"""
        # 1. Merge data sources
        merged = self._merge_sources(static_deps, dynamic_deps)

        # 2. Load previous version
        previous = self._load_previous()

        # 3. Detect changes
        changes = self._detect_changes(previous, merged) if previous else []

        # 4. Save current version
        self._save_current(merged)

        # 5. Generate report
        report = self._generate_report(merged, changes)

        return report

    def _merge_sources(self, static_deps, dynamic_deps):
        """Merge static and dynamic dependency data"""
        all_edges = set()
        edge_data = {}

        for edge in static_deps:
            key = (edge['caller'], edge['callee'])
            all_edges.add(key)
            edge_data[key] = {
                'static': True,
                'dynamic': False,
                'call_count': 0
            }

        for edge in dynamic_deps:
            key = (edge['source'], edge['target'])
            all_edges.add(key)
            if key not in edge_data:
                edge_data[key] = {
                    'static': False,
                    'dynamic': True,
                    'call_count': 0
                }
            edge_data[key]['dynamic'] = True
            edge_data[key]['call_count'] = edge.get('call_count', 0)
            edge_data[key]['p99_latency_ms'] = edge.get('p99_latency_ms')

        return {
            'edges': [
                {'caller': k[0], 'callee': k[1], **v}
                for k, v in edge_data.items()
            ],
            'generated_at': datetime.utcnow().isoformat(),
            'total_edges': len(all_edges)
        }

    def _load_previous(self):
        """Load the previous version of the dependency map"""
        files = sorted(
            f for f in os.listdir(self.storage_path)
            if f.startswith('dep-map-')
        )
        if not files:
            return None

        latest = files[-1]
        path = os.path.join(self.storage_path, latest)
        with open(path) as f:
            return json.load(f)

    def _detect_changes(self, previous, current):
        """Detect dependency graph changes"""
        prev_edges = {
            (e['caller'], e['callee']) for e in previous.get('edges', [])
        }
        curr_edges = {
            (e['caller'], e['callee']) for e in current.get('edges', [])
        }

        added = curr_edges - prev_edges
        removed = prev_edges - curr_edges

        changes = []
        for caller, callee in added:
            changes.append({
                'type': 'added',
                'caller': caller,
                'callee': callee,
                'severity': 'medium'
            })
        for caller, callee in removed:
            changes.append({
                'type': 'removed',
                'caller': caller,
                'callee': callee,
                'severity': 'low'
            })

        return changes

    def _save_current(self, data):
        """Save the current version of the dependency map"""
        timestamp = datetime.utcnow().strftime('%Y%m%d-%H%M%S')
        filename = f'dep-map-{timestamp}.json'
        path = os.path.join(self.storage_path, filename)
        with open(path, 'w') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    def _generate_report(self, data, changes):
        """Generate change report"""
        return {
            'timestamp': data['generated_at'],
            'total_edges': data['total_edges'],
            'total_services': len(set(
                e['caller'] for e in data['edges']
            ) | set(
                e['callee'] for e in data['edges']
            )),
            'changes': changes,
            'new_dependencies': [
                c for c in changes if c['type'] == 'added'
            ],
            'removed_dependencies': [
                c for c in changes if c['type'] == 'removed'
            ],
            'summary': (
                f"Dependency graph updated: {data['total_edges']} edges, "
                f"{len(changes)} changes"
                f" ({sum(1 for c in changes if c['type']=='added')} added, "
                f"{sum(1 for c in changes if c['type']=='removed')} removed)"
            )
        }

# Usage example
if __name__ == '__main__':
    pipeline = DependencyMapPipeline()

    static = [
        {'caller': 'order', 'callee': 'inventory'},
        {'caller': 'order', 'callee': 'payment'},
    ]

    dynamic = [
        {'source': 'order', 'target': 'inventory', 'call_count': 5000},
        {'source': 'order', 'target': 'recommendation', 'call_count': 1000},
    ]

    report = pipeline.run(static, dynamic)
    print(json.dumps(report, indent=2, ensure_ascii=False))

Visualization and Alerting

Dependency maps need visual presentation to deliver value. Grafana with Graphviz or Cytoscape.js are recommended for interactive display:

#!/usr/bin/env python3
"""Generate Graphviz DOT format dependency graph for visualization"""

import json

def generate_dot(dependency_graph, highlight_service=None):
    """
    Convert dependency graph to Graphviz DOT format

    Args:
        dependency_graph: Dependency graph data
        highlight_service: Service to highlight (for failure domain analysis)
    """
    lines = ['digraph service_dependencies {']
    lines.append('  rankdir=LR;')
    lines.append('  fontname="Arial";')
    lines.append('  node [fontname="Arial", shape=box, style=rounded];')
    lines.append('  edge [fontname="Arial"];')
    lines.append('')

    # Node definitions
    nodes = set()
    for edge in dependency_graph.get('edges', []):
        nodes.add(edge['caller'])
        nodes.add(edge['callee'])

    for node in sorted(nodes):
        if node == highlight_service:
            lines.append(
                f'  "{node}" [color=red, style="rounded,filled", '
                f'fillcolor=lightcoral, penwidth=3];'
            )
        else:
            lines.append(f'  "{node}" [style="rounded,filled", '
                        f'fillcolor=lightblue];')

    lines.append('')

    # Edge definitions
    for edge in dependency_graph.get('edges', []):
        caller = edge['caller']
        callee = edge['callee']
        criticality = edge.get('criticality', 'strong')

        if criticality == 'weak':
            color = 'gray'
            style = 'dashed'
            label = 'weak'
        elif edge.get('fallback'):
            color = 'orange'
            style = 'dashed'
            label = 'fallback'
        else:
            color = 'black'
            style = 'solid'
            label = ''

        # Highlight edges related to the failed service
        if highlight_service and (
            caller == highlight_service or callee == highlight_service
        ):
            color = 'red'
            penwidth = '3'
        else:
            penwidth = '1'

        lines.append(
            f'  "{caller}" -> "{callee}" [color={color}, '
            f'style={style}, label="{label}", penwidth={penwidth}];'
        )

    lines.append('}')
    return '\n'.join(lines)

if __name__ == '__main__':
    sample_graph = {
        'edges': [
            {'caller': 'api-gateway', 'callee': 'order-service',
             'criticality': 'strong'},
            {'caller': 'order-service', 'callee': 'inventory-service',
             'criticality': 'strong'},
            {'caller': 'order-service', 'callee': 'payment-service',
             'criticality': 'strong'},
            {'caller': 'order-service', 'callee': 'recommendation-service',
             'criticality': 'weak', 'fallback': True},
            {'caller': 'payment-service', 'callee': 'fraud-detection',
             'criticality': 'strong'},
        ]
    }

    dot = generate_dot(sample_graph, highlight_service='inventory-service')
    print(dot)
    # Generate image with: dot -Tsvg output.dot -o dependency.svg

Production Environment Checklist

Routine Operations Checklist

Check Item	Frequency	Tool	Focus
Dependency map completeness	Daily	Trace analysis + config scanning	Whether dynamically discovered new edges are documented
Failure domain analysis report	Weekly	Custom analysis tool	Top 5 services by blast radius
Single point of failure identification	Weekly	Dependency graph analysis	Strong dependency services without fallback
Architecture change audit	Per deployment	CI/CD pipeline	Whether new/removed dependencies are reviewed
Circuit breaker configuration review	Monthly	Istio/Resilience4j	Whether thresholds are reasonable
Capacity utilization check	Daily	Prometheus	Bottleneck analysis of shared resource dependencies

Common Pitfalls and Corrections

Pitfall 1: The dependency map only needs to be built once

Architecture evolves continuously with new services going live and old ones being decommissioned every week. A dependency map must be continuously updated, otherwise it becomes “an outdated map is worse than no map.”

Correction: Establish an automated pipeline that updates the dependency graph daily from traces and configurations, generating weekly change reports.

Pitfall 2: All dependencies need circuit breakers

Circuit breakers themselves carry complexity and maintenance costs. Over-configuring circuit breakers for low-risk, high-frequency internal service calls can increase false-positive tripping.

Correction: Sort by blast radius score, prioritize circuit breakers for the top 10 services. Use fallback rather than circuit breakers for weak dependencies.

Pitfall 3: Failure domain analysis is a one-time exercise

Failure domains change as system architecture evolves. A service that was well-isolated may become a cross-team single point of failure because a new database connection was added.

Correction: Update failure domain analysis during every architecture review. Integrate dependency change detection in CI/CD, automatically triggering blast radius assessment for new strong dependencies.

Pitfall 4: Only caring about synchronous call dependencies

Asynchronous message queue consumer failures can cause message accumulation, eventually blocking producers. Shared cache failures can trigger cache avalanches across all services depending on that cache.

Correction: The dependency map must include asynchronous dependencies and shared resource dependencies. Monitor consumption lag and backlog for message queues.

Summary

Service dependency maps and failure domain analysis are foundational capabilities of SRE reliability engineering. Without clear dependency awareness, incident troubleshooting relies on luck; without failure domain boundary control, minor incidents can escalate into major disasters at any time.

Key takeaways:

Multi-source fusion discovery: Static configuration scanning ensures coverage completeness, dynamic trace observation ensures accuracy, and the two complement each other to build a trustworthy dependency map
Hierarchical failure domain model: From container to global, each layer has corresponding isolation methods, and nested failure domains form the foundation of system resilience
Quantifiable blast radius: Use graph algorithms to compute failure propagation paths and affected service counts, driving prioritization with data
Combined control strategies: Circuit breakers truncate synchronous call propagation, fallback strategies ensure weak dependencies do not drag down the critical path, and bulkhead isolation prevents resource contention
Continuous updates are key: A dependency map is not a document but living data that must be continuously maintained through automated pipelines

The ultimate goal is not to eliminate all failures — that is unrealistic — but to make the impact scope of every failure controllable, predictable, and quickly recoverable. A system with controllable blast radius is a truly reliable system.

Overview#

The Complexity of Service Dependencies#

Dependency Characteristics in Microservice Architectures#

Dependency Classification System#

Dependency Topology Discovery Methods#

Static Discovery: Extracting from Code and Configuration#

Dynamic Discovery: Runtime Traffic Observation#

Complementary Strategy: Static and Dynamic Discovery#

Failure Domain Analysis#

What Is a Failure Domain#

Hierarchical Model of Failure Domains#

Failure Propagation Path Analysis#

Blast Radius Control Strategies#

Continuous Maintenance of Dependency Maps#

Automated Topology Discovery Pipeline#

Visualization and Alerting#

Production Environment Checklist#

Routine Operations Checklist#

Common Pitfalls and Corrections#

Summary#