Platform Observability and Metrics

📖 Introduction

You can't improve what you don't measure. Platform engineering requires a clear understanding of how the platform impacts developer productivity, system reliability, and business outcomes. The challenge is choosing the right metrics—measuring the wrong things leads to gaming metrics instead of genuine improvement.

In my experience, the most effective platform metrics focus on outcomes (what developers achieve) rather than outputs (what the platform does). DORA metrics have become the industry standard because they correlate strongly with organizational performance.

---

🎯 The Metrics Hierarchy

---

📊 DORA Metrics

The Four Key Metrics

Metric	Definition	Elite	High	Medium	Low
Deployment Frequency	How often code deploys to production	On-demand (multiple/day)	Daily-weekly	Weekly-monthly	Monthly+
Lead Time for Changes	Time from commit to production	< 1 hour	1 day - 1 week	1 week - 1 month	1 month+
Change Failure Rate	% of deployments causing failure	0-15%	16-30%	31-45%	46%+
Time to Restore	Time to recover from failure	< 1 hour	< 1 day	< 1 week	1 week+

DORA Metrics Collection

Python DORA Calculator

"""
DORA metrics calculation and reporting.
"""

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from enum import Enum
import statistics


class PerformanceLevel(Enum):
    ELITE = "elite"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class Deployment:
    """Represents a production deployment."""
    id: str
    team: str
    service: str
    timestamp: datetime
    commit_sha: str
    commit_timestamp: datetime
    success: bool
    caused_incident: bool = False


@dataclass
class Incident:
    """Represents a production incident."""
    id: str
    team: str
    service: str
    started_at: datetime
    resolved_at: Optional[datetime]
    caused_by_deployment: Optional[str] = None  # Deployment ID


@dataclass
class DORAMetrics:
    """DORA metrics for a team."""
    team: str
    period_start: datetime
    period_end: datetime
    
    deployment_frequency: float  # Deployments per day
    lead_time_hours: float  # Hours from commit to deploy
    change_failure_rate: float  # Percentage
    mttr_hours: float  # Mean time to restore in hours
    
    deployment_frequency_level: PerformanceLevel = PerformanceLevel.LOW
    lead_time_level: PerformanceLevel = PerformanceLevel.LOW
    change_failure_level: PerformanceLevel = PerformanceLevel.LOW
    mttr_level: PerformanceLevel = PerformanceLevel.LOW
    
    def __post_init__(self):
        """Calculate performance levels."""
        # Deployment Frequency
        if self.deployment_frequency >= 1:  # Multiple per day
            self.deployment_frequency_level = PerformanceLevel.ELITE
        elif self.deployment_frequency >= 0.14:  # Daily-weekly
            self.deployment_frequency_level = PerformanceLevel.HIGH
        elif self.deployment_frequency >= 0.03:  # Weekly-monthly
            self.deployment_frequency_level = PerformanceLevel.MEDIUM
        else:
            self.deployment_frequency_level = PerformanceLevel.LOW
        
        # Lead Time
        if self.lead_time_hours < 1:
            self.lead_time_level = PerformanceLevel.ELITE
        elif self.lead_time_hours < 24 * 7:  # Less than a week
            self.lead_time_level = PerformanceLevel.HIGH
        elif self.lead_time_hours < 24 * 30:  # Less than a month
            self.lead_time_level = PerformanceLevel.MEDIUM
        else:
            self.lead_time_level = PerformanceLevel.LOW
        
        # Change Failure Rate
        if self.change_failure_rate <= 15:
            self.change_failure_level = PerformanceLevel.ELITE
        elif self.change_failure_rate <= 30:
            self.change_failure_level = PerformanceLevel.HIGH
        elif self.change_failure_rate <= 45:
            self.change_failure_level = PerformanceLevel.MEDIUM
        else:
            self.change_failure_level = PerformanceLevel.LOW
        
        # MTTR
        if self.mttr_hours < 1:
            self.mttr_level = PerformanceLevel.ELITE
        elif self.mttr_hours < 24:
            self.mttr_level = PerformanceLevel.HIGH
        elif self.mttr_hours < 24 * 7:
            self.mttr_level = PerformanceLevel.MEDIUM
        else:
            self.mttr_level = PerformanceLevel.LOW
    
    @property
    def overall_level(self) -> PerformanceLevel:
        """Calculate overall performance level."""
        levels = [
            self.deployment_frequency_level,
            self.lead_time_level,
            self.change_failure_level,
            self.mttr_level,
        ]
        
        level_values = {
            PerformanceLevel.ELITE: 4,
            PerformanceLevel.HIGH: 3,
            PerformanceLevel.MEDIUM: 2,
            PerformanceLevel.LOW: 1,
        }
        
        avg = statistics.mean(level_values[l] for l in levels)
        
        if avg >= 3.5:
            return PerformanceLevel.ELITE
        elif avg >= 2.5:
            return PerformanceLevel.HIGH
        elif avg >= 1.5:
            return PerformanceLevel.MEDIUM
        else:
            return PerformanceLevel.LOW


class DORACalculator:
    """Calculate DORA metrics from deployment and incident data."""
    
    def __init__(self):
        self.deployments: List[Deployment] = []
        self.incidents: List[Incident] = []
    
    def add_deployment(self, deployment: Deployment):
        """Add a deployment to the dataset."""
        self.deployments.append(deployment)
    
    def add_incident(self, incident: Incident):
        """Add an incident to the dataset."""
        self.incidents.append(incident)
    
    def calculate(
        self,
        team: str,
        period_start: datetime,
        period_end: datetime,
    ) -> DORAMetrics:
        """Calculate DORA metrics for a team and period."""
        
        team_deployments = [
            d for d in self.deployments
            if d.team == team
            and period_start <= d.timestamp <= period_end
        ]
        
        team_incidents = [
            i for i in self.incidents
            if i.team == team
            and period_start <= i.started_at <= period_end
        ]
        
        # Deployment Frequency
        days = (period_end - period_start).days or 1
        deployment_frequency = len(team_deployments) / days
        
        # Lead Time for Changes
        lead_times = [
            (d.timestamp - d.commit_timestamp).total_seconds() / 3600
            for d in team_deployments
            if d.commit_timestamp
        ]
        lead_time_hours = statistics.mean(lead_times) if lead_times else 0
        
        # Change Failure Rate
        failed_deployments = sum(1 for d in team_deployments if d.caused_incident)
        change_failure_rate = (
            (failed_deployments / len(team_deployments) * 100)
            if team_deployments else 0
        )
        
        # Mean Time to Restore
        restoration_times = [
            (i.resolved_at - i.started_at).total_seconds() / 3600
            for i in team_incidents
            if i.resolved_at
        ]
        mttr_hours = statistics.mean(restoration_times) if restoration_times else 0
        
        return DORAMetrics(
            team=team,
            period_start=period_start,
            period_end=period_end,
            deployment_frequency=deployment_frequency,
            lead_time_hours=lead_time_hours,
            change_failure_rate=change_failure_rate,
            mttr_hours=mttr_hours,
        )
    
    def generate_report(self, metrics: DORAMetrics) -> str:
        """Generate human-readable report."""
        level_emoji = {
            PerformanceLevel.ELITE: "🏆",
            PerformanceLevel.HIGH: "✅",
            PerformanceLevel.MEDIUM: "⚠️",
            PerformanceLevel.LOW: "❌",
        }
        
        lines = [
            f"DORA Metrics Report: {metrics.team}",
            f"Period: {metrics.period_start.date()} to {metrics.period_end.date()}",
            "=" * 50,
            "",
            f"Overall Performance: {level_emoji[metrics.overall_level]} {metrics.overall_level.value.upper()}",
            "",
            "Metrics:",
            f"  {level_emoji[metrics.deployment_frequency_level]} Deployment Frequency: {metrics.deployment_frequency:.2f}/day",
            f"  {level_emoji[metrics.lead_time_level]} Lead Time: {metrics.lead_time_hours:.1f} hours",
            f"  {level_emoji[metrics.change_failure_level]} Change Failure Rate: {metrics.change_failure_rate:.1f}%",
            f"  {level_emoji[metrics.mttr_level]} MTTR: {metrics.mttr_hours:.1f} hours",
        ]
        
        return "\n".join(lines)


# Example usage
calc = DORACalculator()

# Add sample data
base_time = datetime.now() - timedelta(days=30)

for i in range(60):  # 60 deployments in 30 days = 2/day
    calc.add_deployment(Deployment(
        id=f"deploy-{i}",
        team="checkout",
        service="checkout-api",
        timestamp=base_time + timedelta(hours=i * 12),
        commit_sha=f"abc{i}",
        commit_timestamp=base_time + timedelta(hours=i * 12 - 2),  # 2 hour lead time
        success=True,
        caused_incident=(i % 20 == 0),  # 5% failure rate
    ))

for i in range(3):
    calc.add_incident(Incident(
        id=f"incident-{i}",
        team="checkout",
        service="checkout-api",
        started_at=base_time + timedelta(days=i * 10),
        resolved_at=base_time + timedelta(days=i * 10, hours=1),  # 1 hour MTTR
        caused_by_deployment=f"deploy-{i * 20}",
    ))

metrics = calc.calculate(
    team="checkout",
    period_start=base_time,
    period_end=datetime.now(),
)

print(calc.generate_report(metrics))

---

👩‍💻 Developer Productivity Metrics

SPACE Framework

Developer Experience Survey

"""
Developer experience survey and analysis.
"""

from dataclasses import dataclass
from typing import Dict, List, Optional
from enum import Enum
from datetime import datetime
import statistics


class SurveyCategory(Enum):
    PLATFORM_USABILITY = "platform_usability"
    DOCUMENTATION = "documentation"
    DEPLOYMENT_EXPERIENCE = "deployment_experience"
    SUPPORT_QUALITY = "support_quality"
    SELF_SERVICE = "self_service"
    OVERALL_SATISFACTION = "overall_satisfaction"


@dataclass
class SurveyQuestion:
    """A survey question."""
    id: str
    category: SurveyCategory
    text: str
    scale_min: int = 1
    scale_max: int = 5


@dataclass
class SurveyResponse:
    """A survey response from a developer."""
    respondent_id: str  # Anonymous
    team: str
    timestamp: datetime
    answers: Dict[str, int]  # question_id -> score
    comments: Optional[str] = None


class DeveloperExperienceSurvey:
    """Manage developer experience surveys."""
    
    QUESTIONS = [
        SurveyQuestion(
            "q1", SurveyCategory.PLATFORM_USABILITY,
            "How easy is it to deploy a new service?",
        ),
        SurveyQuestion(
            "q2", SurveyCategory.PLATFORM_USABILITY,
            "How easy is it to find what you need in the developer portal?",
        ),
        SurveyQuestion(
            "q3", SurveyCategory.DOCUMENTATION,
            "How helpful is the platform documentation?",
        ),
        SurveyQuestion(
            "q4", SurveyCategory.DOCUMENTATION,
            "How up-to-date is the documentation?",
        ),
        SurveyQuestion(
            "q5", SurveyCategory.DEPLOYMENT_EXPERIENCE,
            "How confident are you in the deployment process?",
        ),
        SurveyQuestion(
            "q6", SurveyCategory.DEPLOYMENT_EXPERIENCE,
            "How quickly can you get feedback on deployment issues?",
        ),
        SurveyQuestion(
            "q7", SurveyCategory.SUPPORT_QUALITY,
            "How satisfied are you with platform team responsiveness?",
        ),
        SurveyQuestion(
            "q8", SurveyCategory.SUPPORT_QUALITY,
            "How helpful is the platform team in resolving issues?",
        ),
        SurveyQuestion(
            "q9", SurveyCategory.SELF_SERVICE,
            "How often can you complete tasks without asking for help?",
        ),
        SurveyQuestion(
            "q10", SurveyCategory.SELF_SERVICE,
            "How satisfied are you with the available golden paths?",
        ),
        SurveyQuestion(
            "q11", SurveyCategory.OVERALL_SATISFACTION,
            "How likely are you to recommend the platform to a colleague?",
            scale_min=0, scale_max=10,  # NPS scale
        ),
        SurveyQuestion(
            "q12", SurveyCategory.OVERALL_SATISFACTION,
            "Overall, how satisfied are you with the developer platform?",
        ),
    ]
    
    def __init__(self):
        self.responses: List[SurveyResponse] = []
        self.questions_by_id = {q.id: q for q in self.QUESTIONS}
    
    def add_response(self, response: SurveyResponse):
        """Add a survey response."""
        self.responses.append(response)
    
    def calculate_nps(self) -> Dict[str, float]:
        """Calculate Net Promoter Score."""
        nps_question = "q11"
        scores = [
            r.answers[nps_question]
            for r in self.responses
            if nps_question in r.answers
        ]
        
        if not scores:
            return {"nps": 0, "promoters": 0, "passives": 0, "detractors": 0}
        
        promoters = sum(1 for s in scores if s >= 9)
        passives = sum(1 for s in scores if 7 <= s <= 8)
        detractors = sum(1 for s in scores if s <= 6)
        
        total = len(scores)
        nps = ((promoters - detractors) / total) * 100
        
        return {
            "nps": nps,
            "promoters": promoters / total * 100,
            "passives": passives / total * 100,
            "detractors": detractors / total * 100,
        }
    
    def calculate_category_scores(self) -> Dict[SurveyCategory, float]:
        """Calculate average score by category."""
        category_scores: Dict[SurveyCategory, List[float]] = {
            cat: [] for cat in SurveyCategory
        }
        
        for response in self.responses:
            for q_id, score in response.answers.items():
                if q_id in self.questions_by_id:
                    q = self.questions_by_id[q_id]
                    # Normalize to 0-100 scale
                    normalized = (score - q.scale_min) / (q.scale_max - q.scale_min) * 100
                    category_scores[q.category].append(normalized)
        
        return {
            cat: statistics.mean(scores) if scores else 0
            for cat, scores in category_scores.items()
        }
    
    def calculate_team_scores(self) -> Dict[str, float]:
        """Calculate average score by team."""
        team_scores: Dict[str, List[float]] = {}
        
        for response in self.responses:
            if response.team not in team_scores:
                team_scores[response.team] = []
            
            # Calculate average for this response
            scores = list(response.answers.values())
            if scores:
                team_scores[response.team].append(statistics.mean(scores))
        
        return {
            team: statistics.mean(scores)
            for team, scores in team_scores.items()
        }
    
    def identify_improvement_areas(self) -> List[tuple]:
        """Identify lowest scoring areas."""
        question_scores: Dict[str, List[int]] = {}
        
        for response in self.responses:
            for q_id, score in response.answers.items():
                if q_id not in question_scores:
                    question_scores[q_id] = []
                question_scores[q_id].append(score)
        
        averages = [
            (
                q_id,
                self.questions_by_id[q_id].text,
                statistics.mean(scores),
            )
            for q_id, scores in question_scores.items()
            if q_id in self.questions_by_id
        ]
        
        # Sort by score ascending (lowest first)
        return sorted(averages, key=lambda x: x[2])[:5]
    
    def generate_report(self) -> str:
        """Generate survey analysis report."""
        lines = [
            "Developer Experience Survey Report",
            f"Responses: {len(self.responses)}",
            "=" * 50,
            "",
        ]
        
        # NPS
        nps = self.calculate_nps()
        lines.append(f"Net Promoter Score: {nps['nps']:.0f}")
        lines.append(f"  Promoters: {nps['promoters']:.1f}%")
        lines.append(f"  Passives: {nps['passives']:.1f}%")
        lines.append(f"  Detractors: {nps['detractors']:.1f}%")
        lines.append("")
        
        # Category Scores
        lines.append("Category Scores (0-100):")
        for cat, score in self.calculate_category_scores().items():
            emoji = "🟢" if score >= 70 else "🟡" if score >= 50 else "🔴"
            lines.append(f"  {emoji} {cat.value}: {score:.1f}")
        lines.append("")
        
        # Improvement Areas
        lines.append("Top 5 Improvement Areas:")
        for q_id, text, score in self.identify_improvement_areas():
            lines.append(f"  - {text}: {score:.2f}/5")
        
        return "\n".join(lines)


# Example usage
survey = DeveloperExperienceSurvey()

# Add sample responses
for i in range(50):
    survey.add_response(SurveyResponse(
        respondent_id=f"dev-{i}",
        team=["checkout", "payments", "inventory"][i % 3],
        timestamp=datetime.now(),
        answers={
            "q1": 4, "q2": 3, "q3": 4, "q4": 3, "q5": 4,
            "q6": 3, "q7": 4, "q8": 4, "q9": 3, "q10": 4,
            "q11": 8, "q12": 4,
        },
    ))

print(survey.generate_report())

---

📈 Platform Health Metrics

Platform Dashboard

# Grafana dashboard configuration (as JSON model)
apiVersion: 1

dashboards:
  - name: Platform Health
    folder: Platform
    type: file
    disableDeletion: false
    updateIntervalSeconds: 60
    options:
      path: /var/lib/grafana/dashboards

# Key panels for platform dashboard
panels:
  - title: Developer Productivity
    type: stat
    targets:
      - expr: avg(deployment_lead_time_seconds) / 3600
        legendFormat: Lead Time (hours)
      - expr: sum(rate(deployments_total[1d]))
        legendFormat: Deployments/day
  
  - title: Platform Adoption
    type: gauge
    targets:
      - expr: |
          (count(kube_deployment_labels{label_platform_managed="true"}) 
          / count(kube_deployment_labels)) * 100
        legendFormat: Managed Services %
  
  - title: Self-Service Success
    type: timeseries
    targets:
      - expr: |
          sum(rate(self_service_operations_total{status="success"}[1h])) 
          / sum(rate(self_service_operations_total[1h])) * 100
        legendFormat: Success Rate

Metrics Collection Service

"""
Platform metrics collection and aggregation.
"""

from dataclasses import dataclass
from typing import Dict, Any, List
from datetime import datetime, timedelta
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time


# Prometheus metrics
DEPLOYMENT_COUNTER = Counter(
    'platform_deployments_total',
    'Total number of deployments',
    ['team', 'service', 'environment', 'status']
)

DEPLOYMENT_DURATION = Histogram(
    'platform_deployment_duration_seconds',
    'Time taken for deployments',
    ['team', 'environment'],
    buckets=[60, 120, 300, 600, 1200, 1800, 3600]
)

LEAD_TIME = Histogram(
    'platform_lead_time_seconds',
    'Time from commit to deployment',
    ['team'],
    buckets=[300, 900, 1800, 3600, 7200, 14400, 28800, 86400]
)

SELF_SERVICE_OPERATIONS = Counter(
    'platform_self_service_operations_total',
    'Self-service operations',
    ['operation_type', 'status']
)

ACTIVE_SERVICES = Gauge(
    'platform_active_services',
    'Number of active services',
    ['team', 'environment']
)

TEMPLATE_USAGE = Counter(
    'platform_template_usage_total',
    'Template usage count',
    ['template_name', 'team']
)

SUPPORT_TICKETS = Counter(
    'platform_support_tickets_total',
    'Support tickets created',
    ['category', 'priority', 'team']
)

INCIDENT_DURATION = Histogram(
    'platform_incident_duration_seconds',
    'Duration of incidents',
    ['severity', 'team'],
    buckets=[300, 900, 1800, 3600, 7200, 14400, 28800, 86400]
)


class PlatformMetricsCollector:
    """Collect and expose platform metrics."""
    
    def __init__(self, port: int = 9090):
        self.port = port
    
    def start(self):
        """Start metrics server."""
        start_http_server(self.port)
        print(f"Metrics server started on port {self.port}")
    
    def record_deployment(
        self,
        team: str,
        service: str,
        environment: str,
        status: str,
        duration_seconds: float,
        lead_time_seconds: float,
    ):
        """Record a deployment."""
        DEPLOYMENT_COUNTER.labels(
            team=team,
            service=service,
            environment=environment,
            status=status,
        ).inc()
        
        DEPLOYMENT_DURATION.labels(
            team=team,
            environment=environment,
        ).observe(duration_seconds)
        
        if status == "success":
            LEAD_TIME.labels(team=team).observe(lead_time_seconds)
    
    def record_self_service_operation(
        self,
        operation_type: str,
        status: str,
    ):
        """Record a self-service operation."""
        SELF_SERVICE_OPERATIONS.labels(
            operation_type=operation_type,
            status=status,
        ).inc()
    
    def update_active_services(self, counts: Dict[tuple, int]):
        """Update active service counts."""
        for (team, env), count in counts.items():
            ACTIVE_SERVICES.labels(team=team, environment=env).set(count)
    
    def record_template_usage(self, template_name: str, team: str):
        """Record template usage."""
        TEMPLATE_USAGE.labels(
            template_name=template_name,
            team=team,
        ).inc()
    
    def record_support_ticket(
        self,
        category: str,
        priority: str,
        team: str,
    ):
        """Record a support ticket."""
        SUPPORT_TICKETS.labels(
            category=category,
            priority=priority,
            team=team,
        ).inc()
    
    def record_incident(
        self,
        severity: str,
        team: str,
        duration_seconds: float,
    ):
        """Record an incident."""
        INCIDENT_DURATION.labels(
            severity=severity,
            team=team,
        ).observe(duration_seconds)


@dataclass
class PlatformHealthScore:
    """Overall platform health score."""
    
    deployment_success_rate: float
    self_service_success_rate: float
    template_adoption_rate: float
    average_lead_time_hours: float
    incident_rate: float
    developer_satisfaction: float  # From surveys
    
    @property
    def overall_score(self) -> float:
        """Calculate overall health score (0-100)."""
        weights = {
            'deployment': 0.2,
            'self_service': 0.15,
            'adoption': 0.15,
            'lead_time': 0.2,
            'incidents': 0.1,
            'satisfaction': 0.2,
        }
        
        # Normalize each metric to 0-100
        scores = {
            'deployment': self.deployment_success_rate,
            'self_service': self.self_service_success_rate,
            'adoption': self.template_adoption_rate,
            'lead_time': max(0, 100 - (self.average_lead_time_hours * 2)),  # Penalty for long lead times
            'incidents': max(0, 100 - (self.incident_rate * 100)),  # Penalty for incidents
            'satisfaction': self.developer_satisfaction,
        }
        
        return sum(scores[k] * weights[k] for k in weights)
    
    @property
    def status(self) -> str:
        """Get health status."""
        score = self.overall_score
        if score >= 80:
            return "🟢 Healthy"
        elif score >= 60:
            return "🟡 Needs Attention"
        else:
            return "🔴 Critical"


class PlatformHealthMonitor:
    """Monitor overall platform health."""
    
    def __init__(self, metrics_collector: PlatformMetricsCollector):
        self.collector = metrics_collector
    
    def calculate_health(self) -> PlatformHealthScore:
        """Calculate current platform health."""
        # In production, query Prometheus/metrics store
        return PlatformHealthScore(
            deployment_success_rate=98.5,
            self_service_success_rate=92.0,
            template_adoption_rate=78.0,
            average_lead_time_hours=4.5,
            incident_rate=0.02,
            developer_satisfaction=85.0,
        )
    
    def generate_health_report(self) -> str:
        """Generate health report."""
        health = self.calculate_health()
        
        return f"""
Platform Health Report
=====================
Overall Score: {health.overall_score:.1f}/100
Status: {health.status}

Metrics:
  Deployment Success Rate: {health.deployment_success_rate:.1f}%
  Self-Service Success Rate: {health.self_service_success_rate:.1f}%
  Template Adoption Rate: {health.template_adoption_rate:.1f}%
  Average Lead Time: {health.average_lead_time_hours:.1f} hours
  Incident Rate: {health.incident_rate:.2%}
  Developer Satisfaction: {health.developer_satisfaction:.1f}%
"""


# Example usage
if __name__ == "__main__":
    collector = PlatformMetricsCollector()
    # collector.start()  # Start Prometheus metrics endpoint
    
    monitor = PlatformHealthMonitor(collector)
    print(monitor.generate_health_report())

---

🔄 Feedback Loops

Continuous Feedback System

Feedback Aggregator

"""
Aggregate feedback from multiple sources.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Dict, Optional
from enum import Enum
import re
from collections import Counter


class FeedbackSource(Enum):
    SURVEY = "survey"
    TICKET = "ticket"
    SLACK = "slack"
    GITHUB_ISSUE = "github_issue"
    OFFICE_HOURS = "office_hours"


class FeedbackCategory(Enum):
    DOCUMENTATION = "documentation"
    USABILITY = "usability"
    PERFORMANCE = "performance"
    FEATURE_REQUEST = "feature_request"
    BUG = "bug"
    SUPPORT = "support"
    OTHER = "other"


@dataclass
class FeedbackItem:
    """Individual feedback item."""
    id: str
    source: FeedbackSource
    category: FeedbackCategory
    timestamp: datetime
    team: str
    content: str
    sentiment: Optional[float] = None  # -1 to 1
    tags: List[str] = field(default_factory=list)
    actionable: bool = True


@dataclass
class FeedbackInsight:
    """Aggregated insight from feedback."""
    category: FeedbackCategory
    theme: str
    frequency: int
    sample_feedback: List[str]
    suggested_action: str
    priority: str  # high, medium, low


class FeedbackAggregator:
    """Aggregate and analyze feedback."""
    
    CATEGORY_KEYWORDS = {
        FeedbackCategory.DOCUMENTATION: [
            "docs", "documentation", "guide", "readme", "unclear", "confusing",
        ],
        FeedbackCategory.USABILITY: [
            "hard to use", "complicated", "confusing", "intuitive", "ui", "ux",
        ],
        FeedbackCategory.PERFORMANCE: [
            "slow", "fast", "performance", "timeout", "latency",
        ],
        FeedbackCategory.FEATURE_REQUEST: [
            "would be nice", "feature", "request", "want", "need", "add",
        ],
        FeedbackCategory.BUG: [
            "bug", "error", "broken", "doesn't work", "failed", "issue",
        ],
    }
    
    def __init__(self):
        self.feedback: List[FeedbackItem] = []
    
    def add_feedback(self, item: FeedbackItem):
        """Add feedback item."""
        # Auto-categorize if not set
        if item.category == FeedbackCategory.OTHER:
            item.category = self._categorize(item.content)
        
        # Auto-tag
        item.tags = self._extract_tags(item.content)
        
        self.feedback.append(item)
    
    def _categorize(self, content: str) -> FeedbackCategory:
        """Categorize feedback based on content."""
        content_lower = content.lower()
        
        for category, keywords in self.CATEGORY_KEYWORDS.items():
            for keyword in keywords:
                if keyword in content_lower:
                    return category
        
        return FeedbackCategory.OTHER
    
    def _extract_tags(self, content: str) -> List[str]:
        """Extract relevant tags from content."""
        tags = []
        content_lower = content.lower()
        
        # Service/product mentions
        products = ["backstage", "kubernetes", "terraform", "github", "gitlab"]
        for product in products:
            if product in content_lower:
                tags.append(product)
        
        # Workflow mentions
        workflows = ["deployment", "ci/cd", "testing", "monitoring"]
        for workflow in workflows:
            if workflow in content_lower:
                tags.append(workflow)
        
        return list(set(tags))
    
    def get_insights(self, days: int = 30) -> List[FeedbackInsight]:
        """Generate insights from recent feedback."""
        cutoff = datetime.now().replace(hour=0, minute=0) - timedelta(days=days)
        recent = [f for f in self.feedback if f.timestamp >= cutoff]
        
        if not recent:
            return []
        
        insights = []
        
        # Group by category
        by_category: Dict[FeedbackCategory, List[FeedbackItem]] = {}
        for item in recent:
            if item.category not in by_category:
                by_category[item.category] = []
            by_category[item.category].append(item)
        
        # Find themes in each category
        for category, items in by_category.items():
            if len(items) < 3:  # Minimum threshold
                continue
            
            # Find common tags
            all_tags = []
            for item in items:
                all_tags.extend(item.tags)
            common_tags = Counter(all_tags).most_common(3)
            
            theme = common_tags[0][0] if common_tags else category.value
            
            insight = FeedbackInsight(
                category=category,
                theme=theme,
                frequency=len(items),
                sample_feedback=[i.content[:100] for i in items[:3]],
                suggested_action=self._suggest_action(category, theme),
                priority=self._calculate_priority(items),
            )
            insights.append(insight)
        
        return sorted(insights, key=lambda x: x.frequency, reverse=True)
    
    def _suggest_action(self, category: FeedbackCategory, theme: str) -> str:
        """Suggest action based on category and theme."""
        actions = {
            FeedbackCategory.DOCUMENTATION: f"Review and update {theme} documentation",
            FeedbackCategory.USABILITY: f"Conduct UX review of {theme} workflows",
            FeedbackCategory.PERFORMANCE: f"Performance analysis of {theme}",
            FeedbackCategory.FEATURE_REQUEST: f"Add {theme} to backlog for prioritization",
            FeedbackCategory.BUG: f"Investigate and fix {theme} issues",
        }
        return actions.get(category, f"Review {theme} feedback")
    
    def _calculate_priority(self, items: List[FeedbackItem]) -> str:
        """Calculate priority based on frequency and sentiment."""
        if len(items) >= 10:
            return "high"
        elif len(items) >= 5:
            return "medium"
        else:
            return "low"
    
    def generate_report(self, days: int = 30) -> str:
        """Generate feedback report."""
        insights = self.get_insights(days)
        
        lines = [
            f"Platform Feedback Report (Last {days} Days)",
            "=" * 50,
            f"Total Feedback Items: {len(self.feedback)}",
            "",
            "Top Insights:",
        ]
        
        for i, insight in enumerate(insights[:5], 1):
            priority_emoji = {"high": "🔴", "medium": "🟡", "low": "🟢"}[insight.priority]
            lines.append(f"\n{i}. {insight.category.value.title()}: {insight.theme}")
            lines.append(f"   {priority_emoji} Priority: {insight.priority}")
            lines.append(f"   Frequency: {insight.frequency} mentions")
            lines.append(f"   Action: {insight.suggested_action}")
        
        return "\n".join(lines)


from datetime import timedelta

# Example
aggregator = FeedbackAggregator()

# Add sample feedback
samples = [
    ("The deployment docs are outdated", FeedbackSource.SLACK, "checkout"),
    ("CI pipeline is slow for large repos", FeedbackSource.TICKET, "payments"),
    ("Would love to have automatic rollbacks", FeedbackSource.SURVEY, "inventory"),
    ("Backstage search doesn't find our services", FeedbackSource.SLACK, "checkout"),
    ("Need better documentation for golden paths", FeedbackSource.GITHUB_ISSUE, "payments"),
]

for content, source, team in samples:
    aggregator.add_feedback(FeedbackItem(
        id=f"fb-{len(aggregator.feedback)}",
        source=source,
        category=FeedbackCategory.OTHER,
        timestamp=datetime.now() - timedelta(days=1),
        team=team,
        content=content,
    ))

print(aggregator.generate_report())

---

✅ Best Practices

Metrics

1. Measure outcomes, not outputs - Focus on developer impact

2. Trend over absolute - Direction matters more than numbers

3. Team-level visibility - Teams own their metrics

4. Avoid vanity metrics - High numbers that don't matter

5. Balance quantitative with qualitative - Surveys complement metrics

Feedback

1. Multiple channels - Meet developers where they are

2. Close the loop - Show what you did with feedback

3. Regular cadence - Quarterly surveys, weekly analysis

4. Celebrate improvements - Share wins from feedback

5. Safe to criticize - Anonymous options available

---

🔗 What's Next?

In Article 12: Building Platform Teams, we'll explore how to structure and scale platform teams, manage stakeholders, and apply the platform-as-product mindset.

---

📚 References

• DORA Research

• SPACE Framework

• Accelerate (Book)

• Google's DevOps Capabilities

• Developer Experience Survey Best Practices