Run a python server on Kubernetes with a single YAML configuration

Ever wanted a simple server up on K8s without building a docker image or publishing to ECR? Well you can do that by mounting a script into a python image and running it. This post shows you how.

The goal

Define a single YAML configuration for kubernetes that contains a python server image and the code to run on it. We will use this to create a simple health check service that pings our app repeatedly.

The k8s deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ping-test
spec:
  replicas: 1
  selector:
    matchLabels:
      app: python-script
  template:
    metadata:
      labels:
        app: python-script
    spec:
      containers:
        - name: test 
          image: python:3.9-slim
          command: ["sh", "-c", "pip install requests boto3 --disable-pip-version-check --root-user-action=ignore && python /scripts/script.py"]
          volumeMounts:
            - name: python-script-volume
              mountPath: /scripts
          env:
            - name: URL
              value: "https://myapp.com/health"
      volumes:
        - name: python-script-volume
          configMap:
            name: python-script
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: python-script
data:
  script.py: |
    import requests
    import boto3
    import time
    import os
    
    url = os.getenv('URL', default=None)
    print(f'Ping test start: {url}')
    
    
    def put_metric(metric, value=1):
      print(f'[{url}] Put metric {metric} value {value}')
      cloudwatch = boto3.client('cloudwatch', region_name='us-west-2')
      cloudwatch.put_metric_data(
        Namespace='Test',
        MetricData=[{'MetricName': metric, 'Value': value, 'Dimensions':[{'Name':'url', 'Value': url}]}],
      )        
    
    put_metric('ping-test-start')
    while True:
        time.sleep(1)
        start_time = time.time()
        response = requests.get(url)
        end_time = time.time()
        elapsed_time = max(end_time - start_time, 1)
        if response.status_code != 200:
            print(f'>>> Error: Received status code {response.status_code} - {url}')
            put_metric('ping-test-error')
            put_metric('ping-test-time', value=elapsed_time)
        else:
            put_metric('ping-test-success')
            put_metric('ping-test-time', value=elapsed_time)
            print(f'Success: Received status code {response.status_code} - {url}')