Kubernetes ConfigMap Updater

Problem Statement

In modern cloud-native environments, application configurations are frequently stored in Kubernetes ConfigMaps. During automated deployments, feature rollouts, or operational maintenance, engineers often need to dynamically update specific key-value pairs inside a ConfigMap without re-applying large YAML manifests or manually opening interactive editors like kubectl edit.

Manual edits are prone to human error, hard to audit, and difficult to automate in CI/CD pipelines or operational scripts. This script provides a deterministic, programmatic interface to inspect, update, or dynamically create Kubernetes ConfigMaps safely. It handles connection authentications (both in-cluster and local dev environments), atomic updates, and missing resource creation cleanly.

How It Works

The script leverages the official Python kubernetes client library (kubernetes.client) to interact directly with the Kubernetes API server (CoreV1Api).

1. Authentication Discovery: The script first attempts to load in-cluster configuration (for running inside a Kubernetes Pod via ServiceAccount tokens). If that fails, it falls back to reading the local ~/.kube/config file (or the path specified by the KUBECONFIG environment variable).
2. Resource Retrieval: It queries the API server for the target ConfigMap within the specified namespace.
3. Delta Application: If found, it updates or adds the specified key-value pair in the ConfigMap's data dictionary while preserving all other existing entries.
4. Patch/Create Execution: The script sends a strategic merge patch request (patch_namespaced_config_map) to write the changes back atomically. If the ConfigMap does not exist and the --create flag is set, it creates a new ConfigMap resource instead.

Full Script

#!/usr/bin/env python3
"""
Kubernetes ConfigMap Updater

Updates or creates key-value pairs inside a target Kubernetes ConfigMap.
Supports both in-cluster pod execution and local kubeconfig usage.
"""

import argparse
import logging
import os
import sys
from typing import Optional

from kubernetes import client, config
from kubernetes.client.rest import ApiException

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)


def initialize_k8s_client() -> client.CoreV1Api:
    """Initializes and returns the Kubernetes CoreV1Api client.

    Tries in-cluster configuration first; falls back to local kubeconfig.
    """
    try:
        config.load_incluster_config()
        logging.info("Authenticated using in-cluster ServiceAccount.")
    except config.ConfigException:
        try:
            config.load_kube_config()
            logging.info("Authenticated using local kubeconfig.")
        except config.ConfigException as e:
            logging.error("Failed to load Kubernetes configuration: %s", e)
            sys.exit(1)

    return client.CoreV1Api()


def update_configmap(
    api: client.CoreV1Api,
    name: str,
    namespace: str,
    key: str,
    value: str,
    create_if_missing: bool = False,
) -> bool:
    """Updates a single key-value pair in a ConfigMap.

    Args:
        api: An active CoreV1Api client instance.
        name: Name of the ConfigMap.
        namespace: Target Kubernetes namespace.
        key: The configuration key to update or create.
        value: The string value to assign to the key.
        create_if_missing: If True, creates the ConfigMap if it doesn't exist.

    Returns:
        bool: True if operation succeeded, False otherwise.
    """
    try:
        # Fetch the current state of the ConfigMap
        cm = api.read_namespaced_config_map(name=name, namespace=namespace)
        
        # Initialize data dict if the ConfigMap exists but data is empty
        if cm.data is None:
            cm.data = {}

        old_value: Optional[str] = cm.data.get(key)
        
        if old_value == value:
            logging.info("Key '%s' already has value '%s'. No change needed.", key, value)
            return True

        # Update key value pair
        cm.data[key] = value

        # Send patch to API server
        patch_body = {"data": cm.data}
        api.patch_namespaced_config_map(name=name, namespace=namespace, body=patch_body)

        if old_value is None:
            logging.info("Added key '%s' to ConfigMap '%s/%s'.", key, namespace, name)
        else:
            logging.info(
                "Updated key '%s' in ConfigMap '%s/%s' (old: '%s', new: '%s').",
                key, namespace, name, old_value, value
            )
        return True

    except ApiException as e:
        if e.status == 404:
            if create_if_missing:
                logging.info(
                    "ConfigMap '%s/%s' not found. Creating new resource...",
                    namespace, name
                )
                new_cm = client.V1ConfigMap(
                    api_version="v1",
                    kind="ConfigMap",
                    metadata=client.V1ObjectMeta(name=name, namespace=namespace),
                    data={key: value},
                )
                api.create_namespaced_config_map(namespace=namespace, body=new_cm)
                logging.info("Created ConfigMap '%s/%s' with '%s'='%s'.", namespace, name, key, value)
                return True
            else:
                logging.error(
                    "ConfigMap '%s/%s' does not exist. Use --create to create it.",
                    namespace, name
                )
                return False
        elif e.status == 403:
            logging.error("Permission denied (RBAC) accessing '%s/%s': %s", namespace, name, e.reason)
            return False
        else:
            logging.error("Kubernetes API error [%d]: %s", e.status, e.reason)
            return False
    except Exception as e:
        logging.error("Unexpected error updating ConfigMap: %s", e)
        return False


def parse_args() -> argparse.Namespace:
    """Parses command-line arguments."""
    default_namespace = os.environ.get("K8S_NAMESPACE", "default")

    parser = argparse.ArgumentParser(
        description="Safely update or create keys in a Kubernetes ConfigMap."
    )
    parser.add_argument(
        "-c", "--configmap", required=True, help="Name of the target ConfigMap"
    )
    parser.add_argument(
        "-k", "--key", required=True, help="Data key to create or update"
    )
    parser.add_argument(
        "-v", "--value", required=True, help="New string value for the specified key"
    )
    parser.add_argument(
        "-n",
        "--namespace",
        default=default_namespace,
        help=f"Target namespace (default: '{default_namespace}')",
    )
    parser.add_argument(
        "--create",
        action="store_true",
        help="Create the ConfigMap if it does not already exist",
    )

    return parser.parse_args()


def main() -> None:
    args = parse_args()
    api_client = initialize_k8s_client()

    success = update_configmap(
        api=api_client,
        name=args.configmap,
        namespace=args.namespace,
        key=args.key,
        value=args.value,
        create_if_missing=args.create,
    )

    if not success:
        sys.exit(1)


if __name__ == "__main__":
    main()

Configuration

The script accepts runtime arguments alongside environment variable defaults:

Usage Examples

Example 1: Updating an Existing ConfigMap Key

Update the DB_HOST key in the app-config ConfigMap in the default namespace:

python3 k8s_cm.py --configmap app-config --key DB_HOST --value db.internal.example.com

Output:

2026-03-31 10:15:30,102 [INFO] Authenticated using local kubeconfig.
2026-03-31 10:15:30,245 [INFO] Updated key 'DB_HOST' in ConfigMap 'default/app-config' (old: 'localhost', new: 'db.internal.example.com').

Example 2: Target Namespace and Create if Missing

Attempt to update a key in a non-existent ConfigMap under the staging namespace, specifying the --create flag:

python3 k8s_cm.py -c payment-service-config -n staging -k LOG_LEVEL -v DEBUG --create

Output:

2026-03-31 10:17:02,410 [INFO] Authenticated using local kubeconfig.
2026-03-31 10:17:02,512 [INFO] ConfigMap 'staging/payment-service-config' not found. Creating new resource...
2026-03-31 10:17:02,601 [INFO] Created ConfigMap 'staging/payment-service-config' with 'LOG_LEVEL'='DEBUG'.

Example 3: Using Environment Variables in Automation Pipelines

Set target namespace globally via environment variable:

export K8S_NAMESPACE=production
python3 k8s_cm.py -c core-api-config -k FEATURE_NEW_CHECKOUT -v "true"

Output:

2026-03-31 10:20:11,833 [INFO] Authenticated using local kubeconfig.
2026-03-31 10:20:12,012 [INFO] Added key 'FEATURE_NEW_CHECKOUT' to ConfigMap 'production/core-api-config'.

Error Handling

The script handles common failure points cleanly:

• Resource Not Found (404): If the target ConfigMap does not exist and --create is omitted, the script logs an error and exits with code 1 rather than throwing a raw stack trace.
• RBAC Authorization Failures (403): Traps permission failures (e.g., service account lacks get, patch, or create permissions on configmaps) and explicitly alerts the operator.
• Authentication Exceptions: Validates cluster connection prerequisites before executing API commands. If no valid kubeconfig or in-cluster token exists, execution halts immediately.
• Idempotency: If the requested key already holds the target value, no patch API request is executed, saving API calls and avoiding unnecessary cluster events.

Extending Ideas

• Deployment Rollout Trigger: Add logic using AppsV1Api to patch a timestamp annotation (e.g., kubectl.kubernetes.io/restartedAt) on related Deployments so pods automatically pick up updated ConfigMaps.
• Bulk Updates from File: Add a --from-file parameter to read entire JSON or YAML key-value pairs from disk and bulk update the target ConfigMap.
• Dry-Run Mode: Implement a --dry-run flag that previews the diff between the existing ConfigMap data and proposed changes without modifying cluster state.
• Secret Support: Adapt the logic to handle Kubernetes Secret objects by automatically applying Base64 encoding/decoding on values.