Architecture Overview

GreenKube is built on Clean Architecture and Hexagonal Architecture principles, ensuring modularity, testability, and extensibility.

Design Principles

Principle	Implementation
Async-First	All I/O uses Python asyncio for non-blocking execution
Database Agnostic	Repository pattern abstracts storage (PostgreSQL, SQLite, Elasticsearch)
Cloud Agnostic	Supports AWS, GCP, Azure, OVH, Scaleway via mapping files
Resilient	Graceful degradation when data sources are unavailable
Transparent	Clear flagging of estimated vs. measured values
Modular	Each component is independently testable and replaceable

High-Level Architecture

1 Presentation Layer

Web Dashboard SvelteKit

REST API FastAPI

CLI Typer + Rich

2 Business Logic Infrastructure-independent

DataProcessor Orchestrator

Carbon Calculator CO₂e computation

Recommender Analysis engine

BasicEstimator Energy model

Config 12-Factor

3 Adapters (I/O)

→ Input (Collectors)

PrometheusCollector

NodeCollector

PodCollector

HPACollector

OpenCostCollector

ElectricityMapsCollector

BoaviztaCollector

Output (Repositories) →

PostgresRepository

SQLiteRepository

ElasticsearchRepo

NodeRepository

RecommendationRepo

EmbodiedRepository

Project Structure

src/greenkube/
├── __init__.py              # Version
├── api/                     # FastAPI server & endpoints
│   ├── app.py               # Application factory
│   ├── routers/             # Route handlers
│   └── ...
├── cli/                     # Typer CLI commands
│   ├── main.py              # CLI entry point
│   └── ...
├── collectors/              # Input adapters
│   ├── base_collector.py    # Abstract base class
│   ├── base_electricity_provider.py # Provider-agnostic grid intensity interface
│   ├── prometheus_collector.py
│   ├── node_collector.py
│   ├── pod_collector.py
│   ├── opencost_collector.py
│   ├── electricity_maps_collector.py # Default electricity provider
│   ├── wattnet_collector.py # Alternative EU electricity provider
│   ├── hpa_collector.py     # HPA detection
│   ├── pv_collector.py      # Orphaned PersistentVolume detection
│   ├── lb_collector.py      # Orphaned LoadBalancer detection
│   ├── boavizta_collector.py
│   └── discovery/           # Service auto-discovery
├── core/                    # Business logic (no external deps)
│   ├── config.py            # Configuration management
│   ├── factory.py           # Repository & service factory
│   ├── processor.py         # Data pipeline orchestrator
│   ├── collection_orchestrator.py # Async collector runner
│   ├── metric_assembler.py  # Combines data into CombinedMetrics
│   ├── node_zone_mapper.py  # Cloud zone → carbon zone mapping
│   ├── prometheus_resource_mapper.py # Per-pod resource maps
│   ├── cost_normalizer.py   # Cost normalization per step
│   ├── historical_range_processor.py # Day-chunked range queries
│   ├── embodied_emissions_service.py # Boavizta integration
│   ├── calculator.py        # Carbon emission calculator
│   ├── recommender.py              # 11-type optimization analysis
│   ├── savings_attributor.py       # Prorates annual savings to collection periods
│   ├── recommendation_realization.py # Links savings to specific recommendations
│   ├── sustainability_score.py     # 7-dimension 0–100 composite score
│   ├── summary_refresher.py        # Pre-computes KPI cache (metrics_summary)
│   ├── metrics_compressor.py       # Raw → hourly aggregation with retention policy
│   ├── k8s_secret_store.py         # Persists config to Kubernetes Secret
│   ├── aggregator.py               # Metric aggregation helpers
│   ├── scheduler.py                # Collection + cache refresh scheduling
│   ├── db.py                       # Database initialization
│   ├── migrations/                 # Versioned SQL migration scripts
│   ├── k8s_client.py               # Kubernetes client helper
│   ├── health.py                   # Service health check probes
│   └── exceptions.py               # Custom exceptions
├── energy/                  # Energy modeling
│   └── estimator.py         # BasicEstimator (power → energy)
├── data/                    # Static data & profiles
│   ├── datacenter_pue_profiles.py
│   ├── instance_profiles.py
│   ├── provider_power_estimates.csv
│   └── electricity_maps_regions_grid_intensity_default.py
├── models/                  # Pydantic data models
│   ├── metrics.py           # CombinedMetric, CostMetric, etc.
│   ├── node.py              # NodeInfo, NodeZoneContext
│   ├── k8s.py               # Kubernetes-specific models
│   ├── boavizta.py          # Boavizta API response models
│   ├── cli.py               # CLI-specific models
│   └── region_mapping.py    # Cloud region → carbon zone
├── storage/                 # Output adapters (repositories)
│   ├── base_repository.py   # Abstract interfaces (ABC)
│   ├── postgres/            # PostgreSQL implementations
│   │   ├── repository.py            # Metrics, carbon intensity
│   │   ├── node_repository.py       # Node SCD2 snapshots
│   │   ├── recommendation_repository.py
│   │   ├── summary_repository.py    # Pre-computed KPI cache
│   │   ├── timeseries_cache_repository.py
│   │   └── savings_repository.py    # Savings ledger
│   ├── sqlite/              # SQLite implementations (same interface)
│   └── elastic/             # Elasticsearch (optional, not production-ready)
├── reporters/               # Report formatting
├── exporters/               # Data export (CSV, JSON)
└── utils/                   # Utilities

Key Abstractions

Repository Pattern

All storage operations go through abstract base classes defined in storage/base_repository.py:

# storage/base_repository.py (simplified)
class CarbonIntensityRepository(ABC):
    @abstractmethod
    async def get_for_zone_at_time(self, zone: str, timestamp: str) -> float | None: ...

    @abstractmethod
    async def save_history(self, history_data: list, zone: str) -> int: ...

    @abstractmethod
    async def write_combined_metrics(self, metrics: List[CombinedMetric]): ...

    @abstractmethod
    async def read_combined_metrics(self, start_time: datetime, end_time: datetime) -> List[CombinedMetric]: ...

class NodeRepository(ABC):
    @abstractmethod
    async def save_nodes(self, nodes: List[NodeInfo]) -> int: ...

    @abstractmethod
    async def get_snapshots(self, start: datetime, end: datetime) -> List[tuple[str, NodeInfo]]: ...

class RecommendationRepository(ABC):
    @abstractmethod
    async def save_recommendations(self, records: List[RecommendationRecord]) -> int: ...

    @abstractmethod
    async def get_recommendations(self, start: datetime, end: datetime, ...) -> List[RecommendationRecord]: ...

Implementations per backend:

• PostgreSQL — PostgresRepository, PostgresNodeRepository, PostgresRecommendationRepository
• SQLite — SQLiteRepository, SQLiteNodeRepository, SQLiteRecommendationRepository
• Elasticsearch — ElasticsearchRepository, ElasticsearchNodeRepository

Factory Pattern

The factory instantiates the correct implementation based on configuration:

# core/factory.py (simplified)
def get_repository(config: Config) -> CarbonIntensityRepository:
    if config.DB_TYPE == "postgres":
        return PostgresRepository(config.DB_CONNECTION_STRING)
    elif config.DB_TYPE == "sqlite":
        return SQLiteRepository(config.DB_PATH)
    elif config.DB_TYPE == "elasticsearch":
        return ElasticsearchRepository(config.ELASTICSEARCH_HOSTS)

def get_node_repository(config: Config) -> NodeRepository: ...
def get_recommendation_repository(config: Config) -> RecommendationRepository: ...
def get_embodied_repository(config: Config) -> EmbodiedRepository: ...

Collector Pattern

All collectors follow a consistent async interface:

class PrometheusCollector:
    async def collect(self) -> PrometheusMetric:
        """Fetch all metrics concurrently."""
        results = await asyncio.gather(
            self._query_cpu(),
            self._query_memory(),
            self._query_network_rx(),
            self._query_network_tx(),
            self._query_disk_read(),
            self._query_disk_write(),
            self._query_restarts(),
            self._query_node_labels(),
        )
        return PrometheusMetric(...)

Electricity Provider Pattern

Grid carbon-intensity sources implement a dedicated abstraction, BaseElectricityProvider (collectors/base_electricity_provider.py), separate from the generic Collector Pattern above — both ElectricityMapsCollector and WattnetCollector implement it:

# collectors/base_electricity_provider.py (simplified)
class BaseElectricityProvider(ABC):
    @abstractmethod
    async def collect(self, zone: str, target_datetime: datetime | None = None) -> list:
        """Return carbon intensity history for a zone, shaped like the
        Electricity Maps payload so any provider can be stored and consumed
        identically by the rest of the pipeline."""
        ...

    async def close(self): ...  # release HTTP clients / tokens

The active implementation is selected by the factory based on the ELECTRICITY_PROVIDER config (electricity_maps, the default, or wattnet):

# core/factory.py (simplified)
def get_electricity_provider() -> BaseElectricityProvider:
    provider = get_config().ELECTRICITY_PROVIDER
    if provider == "wattnet":
        return WattnetCollector(...)
    return ElectricityMapsCollector(...)

DataProcessor and MetricAssembler depend only on the BaseElectricityProvider interface, so the rest of the pipeline (zone mapping, carbon calculation, storage) is unaware of which provider is active. See the Wattnet guide for the alternative EU provider.

Concurrency Model

GreenKube leverages Python’s asyncio throughout. The DataProcessor runs in four sequential phases, delegating to focused collaborators:

DataProcessor.run()
    │
    ├── Phase 1 (sequential)     ← Node discovery (single K8s API call)
    │     NodeCollector.collect()
    │
    ├── Phase 2 (sequential)     ← Zone resolution (uses Phase 1 data)
    │     NodeZoneMapper.resolve_zones()
    │
    ├── Phase 3 (parallel)       ← External metrics + Boavizta
    │     asyncio.gather(
    │       prometheus_collector.collect(),
    │       opencost_collector.collect(),
    │       pod_collector.collect(),
    │       boavizta.get_profiles()
    │     )
    │
    ├── Phase 4 (sequential)     ← Assembly pipeline
    │     PrometheusResourceMapper  ← Build per-pod resource maps
    │     BasicEstimator            ← CPU → energy estimation
    │     CostNormalizer            ← Normalize costs per step
    │     CarbonCalculator          ← Energy → CO₂e
    │     EmbodiedEmissionsService  ← Apply Boavizta embodied CO₂
    │     MetricAssembler           ← Combine into CombinedMetric
    │
    └── repository.write()        ← Async batch insert

Node collection runs alone in Phase 1 (not concurrently with Prometheus/OpenCost) to prevent Kubernetes API client races. Phases 3 and 4 use the node data collected in Phase 1 instead of performing a second redundant Kubernetes API call.

Testing

1250+ unit and integration tests cover all components:

• pytest with pytest-asyncio for async tests
• respx for HTTP request mocking
• unittest.mock.AsyncMock for async component mocking
• AAA pattern (Arrange, Act, Assert) throughout

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=greenkube --cov-report=html

Tip

The core business logic (src/greenkube/core) has zero dependencies on external infrastructure — it can be tested entirely with mocks and in-memory data.