Struggling to understand sub-handlers

[andrewhharmon]

Hello, I'm try to build an operator that when an instance of my crd gets created, it creates a PVC, waits for the PVC to be bound and the creates a deployment. I have this working in theory, but sometimes I get an unrelated failure in the creating of my deployment. When Kopf retries, it retries creating the pvc as well and then I get a failure that the pvc already exists. I "think" my solution is to use sub-handlers, but I'm failing to get that to work. Any help would be greatly appreciated. Here's the basic code I'm trying so far.

@kopf.on.create('myCRD')
async def create_fn(spec, name, namespace, logger, **kwargs):
    logger.info(f"Spec: {spec}")

    @kopf.subhandler(id='create_deployment')
    async def create_deployment(name, namespace, spec, az, logger):
        .... create deployment code
        return deployment

    @kopf.subhandler(id='create_pvc')
    async def create_persistent_volume_claim(id='create_pvc'):
        ... create pvc code
        return pvc


    def get_persistent_volume(pv_name):
        v1 = client.CoreV1Api()
        pv = v1.read_persistent_volume(name=pv_name)
        return pv

    pvc = await create_persistent_volume_claim()
    pv = get_persistent_volume(pvc.spec.volume_name)
    az = pv.metadata.labels['topology.kubernetes.io/zone']
    deployment = await create_deployment(name, namespace, spec, az, logger)

    return {
        "pvc": pvc.metadata.name,
        "deployment": {
                "name": deployment.metadata.name,
                "resources": spec['resources']
            }
    }

[andrewhharmon]

I ended up trying this approach and I do think it's working. so I guess instead of sub-handlers. Each resource is in its own loop. Are there any pros/cons to this approach?

@kopf.on.create('mrCRD')
async def create_pvc(spec, name, namespace, logger, **kwargs):
    logger.info(f"Creating PVC {name}")
    
@kopf.on.create('mrCRD')
async def create_service(spec, name, namespace, logger, **kwargs):
    logger.info(f"Creating  Service {name}")
   
@kopf.on.create('mrCRD')
async def create_ingress(spec, name, namespace, logger, **kwargs):
    logger.info(f"Creating Ingress {name}")

@kopf.on.create('mrCRD')
async def create_deployment(name, namespace, spec, logger, **kwargs):
    """Wait for PVC binding, get AZ, then create Deployment"""
    logger.info(f"Creating Deployment {name}")
    v1 = client.CoreV1Api()
    apps_v1 = client.AppsV1Api()

    while True:
        pvc = v1.read_namespaced_persistent_volume_claim(name, namespace)
        if pvc.status.phase == "Bound":
            pv_name = pvc.spec.volume_name
            logger.info(f"PVC {name} is bound to PV {pv_name}")

            # Get the AZ from the PV
            pv = v1.read_persistent_volume(pv_name)
            az = pv.metadata.labels.get("topology.kubernetes.io/zone", "unknown")
            logger.info(f"PV {pv_name} is in AZ: {az}")

           create deployment.....
        logger.info(f"⏳PVC not yet bound, sleeping for 3 seconds...")
        await asyncio.sleep(3)