Service Mesh & Governance

Introduction

As microservices scale to dozens or hundreds of services, managing cross-cutting concerns becomes overwhelming. From working with large-scale distributed systems, I've seen how service meshes extract networking logic from application code into infrastructure, and how proper governance prevents organizational chaos.

This article covers service mesh concepts, mTLS security, versioning strategies, and team organization patterns.

What is a Service Mesh?

Service Mesh Components

Component	Responsibility
Data Plane	Sidecar proxies handling traffic
Control Plane	Configuration, policy, telemetry
mTLS	Mutual authentication between services
Load Balancing	Smart traffic distribution
Observability	Automatic tracing and metrics
Traffic Control	Routing, retries, circuit breaking

Service Mesh Concepts

Sidecar Pattern

# Kubernetes pod with sidecar proxy (Istio example)
apiVersion: v1
kind: Pod
metadata:
  name: order-service
  labels:
    app: order-service
spec:
  containers:
    # Application container
    - name: order-service
      image: order-service:1.0.0
      ports:
        - containerPort: 8000
    
    # Sidecar proxy (injected by Istio)
    - name: istio-proxy
      image: istio/proxyv2
      ports:
        - containerPort: 15001  # Outbound
        - containerPort: 15006  # Inbound

Traffic Management

# Istio VirtualService for traffic routing
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    # Route 90% to v1, 10% to v2 (canary)
    - route:
        - destination:
            host: order-service
            subset: v1
          weight: 90
        - destination:
            host: order-service
            subset: v2
          weight: 10
      
      # Retry configuration
      retries:
        attempts: 3
        perTryTimeout: 2s
        retryOn: 5xx,reset,connect-failure
      
      # Timeout
      timeout: 10s

---
# DestinationRule for subsets and load balancing
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service
spec:
  host: order-service
  trafficPolicy:
    loadBalancer:
      simple: LEAST_CONN
    connectionPool:
      http:
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
  subsets:
    - name: v1
      labels:
        version: v1
    - name: v2
      labels:
        version: v2

Mutual TLS (mTLS)

Zero-Trust Networking

# Istio PeerAuthentication for mTLS
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: production
spec:
  mtls:
    mode: STRICT  # Require mTLS for all services

---
# AuthorizationPolicy for service-to-service access
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: order-service-policy
  namespace: production
spec:
  selector:
    matchLabels:
      app: order-service
  rules:
    # Allow gateway to access order service
    - from:
        - source:
            principals: ["cluster.local/ns/production/sa/gateway"]
      to:
        - operation:
            methods: ["GET", "POST", "PUT", "DELETE"]
            paths: ["/api/*"]
    
    # Allow payment service to callback
    - from:
        - source:
            principals: ["cluster.local/ns/production/sa/payment-service"]
      to:
        - operation:
            methods: ["POST"]
            paths: ["/internal/payment-callback"]

Certificate Management

# Application doesn't need to handle TLS
# Service mesh handles it transparently

from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
async def health():
    # This HTTP endpoint is automatically secured via mTLS
    # by the service mesh sidecar
    return {"status": "healthy"}


# When calling other services, no TLS code needed
import httpx

async def call_payment_service(order_id: str):
    # Plain HTTP - sidecar encrypts it
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://payment-service:8000/payments",
            json={"order_id": order_id},
        )
        return response.json()

API Versioning Strategies

URL Path Versioning

from fastapi import FastAPI, APIRouter

app = FastAPI()

# Version 1 API
v1_router = APIRouter(prefix="/api/v1")

@v1_router.get("/orders/{order_id}")
async def get_order_v1(order_id: str):
    """V1: Returns basic order info."""
    order = await order_repo.get(order_id)
    return {
        "id": order.id,
        "status": order.status,
        "total": order.total,
    }


# Version 2 API (breaking change)
v2_router = APIRouter(prefix="/api/v2")

@v2_router.get("/orders/{order_id}")
async def get_order_v2(order_id: str):
    """V2: Returns detailed order with items."""
    order = await order_repo.get_with_items(order_id)
    return {
        "id": order.id,
        "status": order.status,
        "total": {
            "amount": order.total,
            "currency": order.currency,
        },
        "items": [
            {"product_id": i.product_id, "quantity": i.quantity}
            for i in order.items
        ],
        "created_at": order.created_at.isoformat(),
    }


app.include_router(v1_router)
app.include_router(v2_router)

Header-Based Versioning

from fastapi import Header, HTTPException


async def get_api_version(
    api_version: str = Header(default="1", alias="X-API-Version")
) -> int:
    """Extract API version from header."""
    try:
        version = int(api_version)
        if version not in [1, 2]:
            raise ValueError()
        return version
    except ValueError:
        raise HTTPException(400, "Invalid API version")


@app.get("/orders/{order_id}")
async def get_order(
    order_id: str,
    version: int = Depends(get_api_version),
):
    """Version-aware endpoint."""
    if version == 1:
        return await get_order_v1(order_id)
    else:
        return await get_order_v2(order_id)

Service Mesh Version Routing

# Route based on header
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: order-service
spec:
  hosts:
    - order-service
  http:
    # Route v2 requests to v2 deployment
    - match:
        - headers:
            x-api-version:
              exact: "2"
      route:
        - destination:
            host: order-service
            subset: v2
    
    # Default to v1
    - route:
        - destination:
            host: order-service
            subset: v1

Contract Testing

# Consumer-driven contract test example
# Using Pact for Python

from pact import Consumer, Provider
import pytest


@pytest.fixture
def pact():
    """Set up Pact consumer."""
    pact = Consumer("order-service").has_pact_with(
        Provider("payment-service"),
        pact_dir="./pacts",
    )
    pact.start_service()
    yield pact
    pact.stop_service()


def test_payment_creation_contract(pact):
    """Test contract: Order service expects this from Payment service."""
    expected_response = {
        "payment_id": "pay_123",
        "status": "pending",
        "amount": 99.99,
    }
    
    pact.given(
        "a valid payment request"
    ).upon_receiving(
        "a request to create payment"
    ).with_request(
        method="POST",
        path="/payments",
        headers={"Content-Type": "application/json"},
        body={
            "order_id": "order_456",
            "amount": 99.99,
            "currency": "USD",
        },
    ).will_respond_with(
        status=201,
        headers={"Content-Type": "application/json"},
        body=expected_response,
    )
    
    with pact:
        # Make request to mock server
        result = payment_client.create_payment(
            order_id="order_456",
            amount=99.99,
        )
        
        assert result["payment_id"] == "pay_123"
        assert result["status"] == "pending"

Backward Compatibility

from pydantic import BaseModel, Field
from typing import Optional


# Version 1 schema
class OrderV1(BaseModel):
    id: str
    status: str
    total: float


# Version 2 schema (backward compatible additions)
class OrderV2(BaseModel):
    id: str
    status: str
    total: float
    # New optional fields (backward compatible)
    currency: str = "USD"
    items: list[dict] = Field(default_factory=list)
    created_at: Optional[str] = None


# Response adapter
class OrderResponseAdapter:
    """Adapt internal order to versioned response."""
    
    @staticmethod
    def to_v1(order: Order) -> OrderV1:
        return OrderV1(
            id=order.id,
            status=order.status,
            total=order.total,
        )
    
    @staticmethod
    def to_v2(order: Order) -> OrderV2:
        return OrderV2(
            id=order.id,
            status=order.status,
            total=order.total,
            currency=order.currency,
            items=[item.to_dict() for item in order.items],
            created_at=order.created_at.isoformat(),
        )


# Deprecation warnings
import warnings
from functools import wraps


def deprecated(version: str, removal_date: str, alternative: str = None):
    """Mark endpoint as deprecated."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            warnings.warn(
                f"{func.__name__} is deprecated since {version} "
                f"and will be removed on {removal_date}. "
                f"Use {alternative} instead." if alternative else "",
                DeprecationWarning,
            )
            response = await func(*args, **kwargs)
            # Add deprecation header
            return response
        
        # Add OpenAPI deprecation
        wrapper.__deprecated__ = True
        return wrapper
    return decorator


@app.get("/api/v1/orders/{order_id}")
@deprecated(
    version="2024-01",
    removal_date="2024-07-01",
    alternative="/api/v2/orders/{order_id}",
)
async def get_order_v1(order_id: str):
    """Deprecated: Use v2 API."""
    pass

Team Organization

Conway's Law in Practice

Service Ownership

# service-catalog.yml
services:
  order-service:
    owner: order-team
    slack: "#order-team-alerts"
    pagerduty: order-team-oncall
    repository: github.com/company/order-service
    documentation: https://docs.internal/order-service
    sla:
      availability: 99.9%
      latency_p99: 200ms
    dependencies:
      - payment-service
      - inventory-service
      - user-service
    
  payment-service:
    owner: payment-team
    slack: "#payment-team-alerts"
    pagerduty: payment-team-oncall
    repository: github.com/company/payment-service
    documentation: https://docs.internal/payment-service
    sla:
      availability: 99.95%
      latency_p99: 500ms
    dependencies:
      - stripe-integration
      - notification-service

API Governance

# Shared API standards library
# company_api_standards/

from fastapi import Request, HTTPException
from pydantic import BaseModel
from typing import TypeVar, Generic
from datetime import datetime


T = TypeVar("T")


class StandardResponse(BaseModel, Generic[T]):
    """Standard API response envelope."""
    success: bool
    data: T | None = None
    error: dict | None = None
    meta: dict = {}
    timestamp: str = datetime.utcnow().isoformat()


class StandardError(BaseModel):
    """Standard error response."""
    code: str
    message: str
    details: dict | None = None


# Middleware for request ID
async def request_id_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response


# Standard exception handler
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=StandardResponse(
            success=False,
            error={
                "code": f"HTTP_{exc.status_code}",
                "message": exc.detail,
            },
        ).model_dump(),
    )


# Health check standard
class HealthCheck(BaseModel):
    status: str
    version: str
    dependencies: dict[str, str]


@app.get("/health", response_model=HealthCheck)
async def health():
    """Standard health check endpoint."""
    return HealthCheck(
        status="healthy",
        version="1.0.0",
        dependencies={
            "database": "healthy",
            "redis": "healthy",
        },
    )

Documentation and Discovery

# API documentation standards
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def custom_openapi(app: FastAPI):
    """Customize OpenAPI schema with company standards."""
    if app.openapi_schema:
        return app.openapi_schema
    
    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )
    
    # Add company-wide info
    openapi_schema["info"]["x-team"] = "order-team"
    openapi_schema["info"]["x-slack"] = "#order-team"
    openapi_schema["info"]["x-sla"] = {
        "availability": "99.9%",
        "latency_p99": "200ms",
    }
    
    # Add server info
    openapi_schema["servers"] = [
        {"url": "https://api.company.com", "description": "Production"},
        {"url": "https://staging-api.company.com", "description": "Staging"},
    ]
    
    app.openapi_schema = openapi_schema
    return app.openapi_schema


app.openapi = lambda: custom_openapi(app)

Service Registry

from dataclasses import dataclass
from typing import Optional


@dataclass
class ServiceInfo:
    name: str
    version: str
    owner: str
    endpoints: list[str]
    health_endpoint: str
    documentation_url: str
    repository: str


class ServiceRegistry:
    """Central service registry."""
    
    def __init__(self, consul_client):
        self.consul = consul_client
    
    async def register(self, service: ServiceInfo):
        """Register service in discovery."""
        await self.consul.agent.service.register(
            name=service.name,
            service_id=f"{service.name}-{uuid.uuid4().hex[:8]}",
            address=os.getenv("SERVICE_HOST"),
            port=int(os.getenv("SERVICE_PORT", 8000)),
            tags=[
                f"version={service.version}",
                f"owner={service.owner}",
            ],
            meta={
                "docs": service.documentation_url,
                "repo": service.repository,
            },
            check={
                "http": f"http://localhost:{os.getenv('SERVICE_PORT', 8000)}{service.health_endpoint}",
                "interval": "10s",
            },
        )
    
    async def discover(self, service_name: str) -> list[dict]:
        """Discover service instances."""
        _, services = await self.consul.health.service(service_name, passing=True)
        return [
            {
                "id": s["Service"]["ID"],
                "address": s["Service"]["Address"],
                "port": s["Service"]["Port"],
                "meta": s["Service"]["Meta"],
            }
            for s in services
        ]

Practical Checklist

Service Readiness Checklist

## Service Production Readiness

### Code Quality
- [ ] Code reviewed and approved
- [ ] Unit tests with >80% coverage
- [ ] Integration tests passing
- [ ] Contract tests defined

### API Design
- [ ] OpenAPI documentation complete
- [ ] Versioning strategy defined
- [ ] Backward compatibility verified
- [ ] Error handling standardized

### Observability
- [ ] Structured logging implemented
- [ ] Metrics exposed (RED metrics)
- [ ] Distributed tracing enabled
- [ ] Health check endpoints

### Resilience
- [ ] Timeouts configured
- [ ] Retries with backoff
- [ ] Circuit breakers implemented
- [ ] Graceful degradation tested

### Security
- [ ] Authentication required
- [ ] Authorization policies defined
- [ ] Secrets managed securely
- [ ] Input validation complete

### Operations
- [ ] Runbook documented
- [ ] Alerts configured
- [ ] On-call rotation assigned
- [ ] Deployment pipeline tested

Key Takeaways

1. Service mesh simplifies - Extract networking from app code

2. mTLS by default - Zero-trust networking

3. Version thoughtfully - Plan for backward compatibility

4. Own your services - Clear team ownership

5. Standardize APIs - Consistent patterns across services

Series Conclusion

This completes our Microservice Architecture 101 series. We've covered:

1. Foundations - When to use microservices

2. Design - Service decomposition and API design

3. Communication - Sync and async patterns

4. Data - Database per service and distributed transactions

5. Operations - Resilience, observability, and deployment

The key to successful microservices is starting simple, measuring everything, and evolving based on real needs rather than theoretical concerns.

---

This article is part of the Microservice Architecture 101 series.