Protect your Kubernetes Persistent Volumes from accidental deletion.

Hello, I would like to specify that the default value in the Storage Class used to provision Persistent Volumes is "Delete" which means that if for some reason the Persistent Volume Claim is removed the same will happen with your Persistent Volume as a result the data will be lost.

So, I would like to strongly suggest to either use the following Python Script to automate the patching the Reclaim Policy of the Persistent Volumes or patch it manually by the following helpful guide found in the official Kubernetes Documentation.

The Python Script is capable of retrieving all the Persistent Volumes in any Kubernetes Namespace and update the PV Reclaim Policy, actually as you can see, it can report which Persistent Volumes are having the opposite value of the desired value you want to be applied as Reclaim Policy (pv_patch_operation).

from kubernetes import client, config
from datetime import datetime
import logging
from logging.handlers import RotatingFileHandler
from pytz import timezone

def timetz(*args):
    return datetime.now(tz).timetuple()

tz = timezone('Europe/Athens') # UTC, Europe/Berlin

logging.Formatter.converter = timetz

logging.basicConfig(
        level=logging.INFO,
        format='[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s',
        datefmt='%Y-%m-%dT%H:%M:%S',
        handlers=[
            RotatingFileHandler('logs/app.log', mode='w', maxBytes=100000000, backupCount=10),
            logging.StreamHandler()
            ]
        )

logging.info('Timezone: ' + str(tz))
logger = logging.getLogger(__name__)  

pv_patch_operation="Retain"

def get_pvs_from_api(v1):

    get_pvs_list=[]
    pv_response = v1.list_persistent_volume(watch=False)

    for pv in pv_response.items:
        get_pvs_list.append({
            "pv_name": pv.metadata.name,
            "pv_retain_policy": pv.spec.persistent_volume_reclaim_policy,
            "pvc_reference": {
                "kind": pv.spec.claim_ref.kind, 
                "deploment_name": pv.spec.claim_ref.name,
                "namespace":  pv.spec.claim_ref.namespace
            } 
        })
    return get_pvs_list

def update_pv_reclaim_policy(v1,get_pvs_list):
    for pvs in get_pvs_list:
        if pvs.get('pv_retain_policy') !=pv_patch_operation: 
            v1.patch_persistent_volume(name=pvs.get('pv_name'), pretty=True, body={'spec': {'persistentVolumeReclaimPolicy': pv_patch_operation }})


def main():
    try: 
        ## Read The kubeconfig and show the current context
        contexts, active_context = config.list_kube_config_contexts()
        if not contexts:
            print("Cannot find any context in kube-config file.")
            return
        contexts = [context['name'] for context in contexts]
        config.load_kube_config(context=active_context['name'])
        logger.info("Current Context: "  + str(active_context['name']))

        ACTION = input("Please Specify Action [get/update]: ")
        logger.info('You are going to "' + str(ACTION) + '" all the pvs from the cluster "' + str(active_context['name']) + '"' )
        CONFIRM = input("Do you want to Proceed[y/n]: ")
        if CONFIRM.strip() in ['y','Y']:
            v1 = client.CoreV1Api()
            get_pvs_list=get_pvs_from_api(v1)
            if len(get_pvs_list) != 0:
                if ACTION.strip() not in ['get','g', 'u','update']:
                    logger.error('Please select one of following options ["g","get","u","update"]')
                    exit(1)
                if ACTION.strip() in ['u','update']:
                    update_pv_reclaim_policy(v1, get_pvs_list)
                    get_pvs_list=get_pvs_from_api(v1)
                    logger.info(get_pvs_list)
                if ACTION.strip() in ['g','get']:
                    logger.info(get_pvs_list)
            else: 
                logger.error('I cannot find any persistent volumes')
        elif CONFIRM in ['n','N']:
            logger.info('Bye.')
            exit(0)
        else: 
            logger.error('Please select between the following valid options ["y","Y","n","N"]')
            exit(1)



    except Exception as e:
        print(e)

if __name__ == "__main__":
    main()

I hope you like the tutorial, if you do give a thumps up! and follow me in Twitter, also you can subscribe to my Newsletter in order to avoid missing any of the upcoming tutorials.

Media Attribution

I would like to thank Clark Tibbs for designing the awesome photo I am using in my posts.