Run a python server on Kubernetes with a single YAML configuration
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}')