Auto-Provisioning Pattern: Resources Created on Demand

“

Series: S10 Design Patterns · Article 13 | Level: Advanced | Reading Time: 18 min

#TL;DR

The Auto-Provisioning Pattern lets coomia-dip automatically create underlying infrastructure resources when users define Ontology Schemas — storage tables, indexes, reasoning engine instances, monitoring dashboards, and more — without manual operations.
coomia-dip implements fully automated provisioning through a four-stage pipeline: Schema Change Detection, Resource Requirement Derivation, Resource Orchestration, and Health Checking.
Supports scale-to-zero, resource templates, and multi-environment provisioning (Dev/Staging/Prod), delivering a true "define and deploy" experience.

#Introduction: Ontology as Infrastructure

In traditional platforms, defining a business object requires extensive operations work:

Code

1. Create database tables and indexes
2. Configure data sync pipelines
3. Set up permissions and access control
4. Deploy reasoning engine instances
5. Configure monitoring and alerts
6. Generate API documentation and SDK

In coomia-dip, all of this happens automatically. Users only need to define the ObjectType Schema, and the platform automatically derives and creates all necessary infrastructure resources.

#Part 1: Auto-Provisioning Pipeline

#1.1 Schema Change Detection

Python

from dataclasses import dataclass, field
from typing import Any
from enum import Enum

class ChangeType(Enum):
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"

@dataclass
class SchemaChange:
    change_type: ChangeType
    object_type: str
    old_schema: dict | None = None
    new_schema: dict | None = None
    diff: dict = field(default_factory=dict)

class SchemaChangeDetector:
    async def detect_changes(
        self, old_schemas: dict[str, dict], new_schemas: dict[str, dict]
    ) -> list[SchemaChange]:
        changes = []
        for name in set(new_schemas) - set(old_schemas):
            changes.append(SchemaChange(
                change_type=ChangeType.CREATE, object_type=name,
                new_schema=new_schemas[name],
            ))
        for name in set(old_schemas) - set(new_schemas):
            changes.append(SchemaChange(
                change_type=ChangeType.DELETE, object_type=name,
                old_schema=old_schemas[name],
            ))
        for name in set(old_schemas) & set(new_schemas):
            if old_schemas[name] != new_schemas[name]:
                changes.append(SchemaChange(
                    change_type=ChangeType.UPDATE, object_type=name,
                    old_schema=old_schemas[name], new_schema=new_schemas[name],
                    diff=self._compute_diff(old_schemas[name], new_schemas[name]),
                ))
        return changes

#1.2 Resource Requirement Derivation

Python

@dataclass
class ResourceRequirement:
    resource_type: str
    Layer: str
    config: dict[str, Any]
    dependencies: list[str] = field(default_factory=list)
    priority: int = 0

class ResourceDeriver:
    async def derive(self, change: SchemaChange) -> list[ResourceRequirement]:
        requirements = []
        if change.change_type == ChangeType.CREATE:
            schema = change.new_schema

            # 1. Iceberg table
            requirements.append(ResourceRequirement(
                resource_type="iceberg_table", Layer="data-Layer",
                config={
                    "table_name": self._to_table_name(change.object_type),
                    "schema": self._to_iceberg_schema(schema),
                    "partition_spec": self._derive_partition(schema),
                },
                priority=0,
            ))

            # 2. Search index
            indexed_fields = self._find_indexed_fields(schema)
            if indexed_fields:
                requirements.append(ResourceRequirement(
                    resource_type="search_index", Layer="data-Layer",
                    config={"index_name": f"idx_{change.object_type.lower()}", "fields": indexed_fields},
                    dependencies=[f"iceberg_table:{change.object_type}"],
                    priority=1,
                ))

            # 3. gRPC endpoint
            requirements.append(ResourceRequirement(
                resource_type="grpc_endpoint", Layer="control-Layer",
                config={"service": "ObjectQueryService", "object_type": change.object_type},
                priority=1,
            ))

            # 4. Reasoning instance (if rules/state machine present)
            if "state_machine" in schema or "rules" in schema:
                requirements.append(ResourceRequirement(
                    resource_type="reasoning_instance", Layer="intelligence-Layer",
                    config={"object_type": change.object_type, "rules": schema.get("rules", [])},
                    priority=2,
                ))

            # 5. Monitoring dashboard
            requirements.append(ResourceRequirement(
                resource_type="monitoring_dashboard", Layer="deployment-Layer",
                config={"object_type": change.object_type},
                priority=3,
            ))

            # 6. SDK type generation
            requirements.append(ResourceRequirement(
                resource_type="sdk_type_generation", Layer="sdk-Layer",
                config={"object_type": change.object_type, "schema": schema, "languages": ["python", "typescript"]},
                priority=4,
            ))

        return requirements

#1.3 Resource Orchestration

Python

class ResourceOrchestrator:
    async def provision(self, requirements: list[ResourceRequirement], context: dict) -> dict:
        ordered = self._topological_sort(requirements)
        results = {}
        for req in ordered:
            try:
                provisioner = self._provisioners[req.resource_type]
                result = await provisioner.provision(req.config, context)
                results[f"{req.resource_type}:{req.config.get('object_type', '')}"] = {
                    "status": "provisioned", "details": result,
                }
            except Exception as e:
                results[f"{req.resource_type}:{req.config.get('object_type', '')}"] = {
                    "status": "failed", "error": str(e),
                }
                if req.priority == 0:
                    raise ProvisioningError(f"Critical resource failed: {req.resource_type}: {e}")
        return results

#Part 2: Resource Templates

#2.1 Predefined Templates

Python

class ResourceTemplate:
    TEMPLATES = {
        "standard_entity": {
            "description": "Standard business entity with CRUD + search",
            "resources": [
                {"type": "iceberg_table", "Layer": "data-Layer"},
                {"type": "search_index", "Layer": "data-Layer"},
                {"type": "grpc_endpoint", "Layer": "control-Layer"},
                {"type": "monitoring_dashboard", "Layer": "deployment-Layer"},
                {"type": "sdk_type_generation", "Layer": "sdk-Layer"},
            ],
        },
        "rule_entity": {
            "description": "Entity with reasoning rules and state machine",
            "resources": [
                {"type": "iceberg_table", "Layer": "data-Layer"},
                {"type": "reasoning_instance", "Layer": "intelligence-Layer"},
                {"type": "agent_monitor", "Layer": "agent-runtime"},
                {"type": "monitoring_dashboard", "Layer": "deployment-Layer"},
            ],
        },
        "event_entity": {
            "description": "Append-only event/log entity",
            "resources": [
                {"type": "iceberg_table", "Layer": "data-Layer",
                 "config_override": {"partition": "day", "sort": "timestamp"}},
            ],
        },
    }

#2.2 Custom Templates

Python

class CustomTemplateRegistry:
    async def register_template(self, name: str, template: dict, tenant_id: str) -> None:
        await self._validate_template(template)
        await self._store.save({
            "name": name, "template": template, "tenant_id": tenant_id,
            "created_at": datetime.utcnow().isoformat(),
        })

    async def apply_template(
        self, template_name: str, object_type: str, schema: dict, context: dict
    ) -> dict:
        template = await self._store.get(template_name)
        requirements = self._template_to_requirements(template, object_type, schema)
        return await self._orchestrator.provision(requirements, context)

#Part 3: Scale-to-Zero

#3.1 Automatic Scale Down

Python

class ScaleToZeroManager:
    async def check_and_scale(self) -> list[dict]:
        actions = []
        for resource in await self._resource_store.list_active():
            last_activity = await self._activity_tracker.get_last_activity(resource.resource_id)
            idle_hours = (datetime.utcnow() - last_activity).total_seconds() / 3600

            if idle_hours > resource.idle_threshold_hours:
                await self._scale_to_zero(resource)
                actions.append({
                    "resource_id": resource.resource_id,
                    "action": "scaled_to_zero", "idle_hours": idle_hours,
                })
        return actions

    async def scale_up_on_demand(self, resource_id: str) -> None:
        resource = await self._resource_store.get(resource_id)
        if resource.status == "scaled_to_zero":
            await self._provisioners[resource.resource_type].resume(resource)
            resource.status = "active"
            await self._resource_store.save(resource)

#3.2 Auto Horizontal Scaling

Python

class AutoScaler:
    async def evaluate(self, resource_id: str) -> dict:
        metrics = await self._metrics_service.get_resource_metrics(resource_id)
        action = None
        if metrics["cpu_utilization"] > 80 or metrics["queue_depth"] > 100:
            action = "scale_up"
        elif metrics["cpu_utilization"] < 20 and metrics["queue_depth"] < 5:
            action = "scale_down"
        if action:
            await self._execute_scaling(resource_id, action)
        return {"resource_id": resource_id, "metrics": metrics, "action": action}

#Part 4: Multi-Environment Provisioning

#4.1 Environment-Aware Provisioning

Python

class EnvironmentAwareProvisioner:
    ENV_CONFIGS = {
        "dev": {
            "iceberg_table": {"replicas": 1, "storage_class": "standard"},
            "reasoning_instance": {"cpu": "0.5", "memory": "512Mi"},
            "monitoring_dashboard": {"enabled": False},
        },
        "staging": {
            "iceberg_table": {"replicas": 2, "storage_class": "standard"},
            "reasoning_instance": {"cpu": "1", "memory": "1Gi"},
        },
        "production": {
            "iceberg_table": {"replicas": 3, "storage_class": "ssd"},
            "reasoning_instance": {"cpu": "4", "memory": "8Gi"},
        },
    }

    async def provision(self, requirement, environment: str, context: dict) -> dict:
        env_config = self.ENV_CONFIGS.get(environment, self.ENV_CONFIGS["dev"])
        merged = {**requirement.config, **env_config.get(requirement.resource_type, {})}
        return await self._provisioners[requirement.resource_type].provision(merged, context)

#Part 5: Health Checking and Self-Healing

Python

class ResourceHealthChecker:
    async def check_all(self, tenant_id: str) -> dict:
        resources = await self._resource_store.list_by_tenant(tenant_id)
        results = {"healthy": 0, "unhealthy": 0, "details": []}

        for resource in resources:
            healthy = await self._check_resource(resource)
            if healthy:
                results["healthy"] += 1
            else:
                results["unhealthy"] += 1
                await self._attempt_self_heal(resource)

            results["details"].append({
                "resource_id": resource.resource_id,
                "type": resource.resource_type, "healthy": healthy,
            })
        return results

    async def _attempt_self_heal(self, resource) -> bool:
        try:
            await self._provisioners[resource.resource_type].recreate(resource.config)
            return True
        except Exception:
            await self._alert_service.send(
                severity="critical",
                title=f"Resource self-heal failed: {resource.resource_id}",
            )
            return False

#Part 6: Resource Cleanup

Python

class ResourceCleanup:
    async def cleanup(self, object_type: str, context: dict) -> dict:
        resources = await self._resource_store.find_by_object_type(object_type)
        cleaned = []
        for resource in reversed(resources):
            try:
                await self._provisioners[resource.resource_type].deprovision(resource.config)
                cleaned.append(resource.resource_id)
            except Exception as e:
                await self._alert_service.send(
                    severity="warning",
                    title=f"Resource cleanup failed: {resource.resource_id}",
                    message=str(e),
                )
        return {"cleaned": len(cleaned), "total": len(resources)}

#Key Takeaways

Define and Deploy: Users only need to define Ontology Schemas — all underlying resources are created automatically
Four-Stage Pipeline: Detect changes, derive requirements, orchestrate provisioning, health check
Resource Templates: Predefined and custom templates accelerate provisioning for common patterns
Scale-to-Zero: Inactive resources automatically scale down and resume on demand
Environment-Aware: Dev/Staging/Prod environments automatically adapt resource specifications
Self-Healing: Health checks + automatic recreation ensure continuous resource availability

#Next Article

S10-14: Hierarchical Multi-Tenancy: Organization-Aware Resource Isolation

#Tags

#DesignPatterns #AutoProvisioning #SchemaDriven #ScaleToZero #ResourceTemplates #MultiEnvironment #SelfHealing