DEV Community

Cover image for CDK8S Example
πŸš€ Vu Dao πŸš€
πŸš€ Vu Dao πŸš€

Posted on • Edited on

CDK8S Example

CDK8s is a software development framework for defining Kubernetes applications and reusable abstractions using familiar programming languages and rich object-oriented APIs. CDK8s generates pure Kubernetes YAML - you can use CDK8s to define applications for any Kubernetes cluster running anywhere.


CDK8S Example

This post provides you some examples of cdk8s python code to create k8s deployments, services, readiness and livness probes, statefulset, persitent volume and cronjob


CDK8S Example


What’s In This Document


πŸš€ Getting Started With cdk8s

Getting Started
1. Prerequisites

  • Python >= 3.7.7
  • pipenv version 2018.11.26 or above.

2. New Project

 $ mkdir hello $ cd hello $ cdk8s init python-app Initializing a project from the python-app template 
Enter fullscreen mode Exit fullscreen mode

πŸš€ Create cluster IP deployment and target port service

 from constructs import Construct from imports import k8s class ClusterIp(Construct): def __init__(self, scope: Construct, id: str): super().__init__(scope, id) app_name = 'wappip' label = {'app': app_name} k8s.KubeDeployment( self, "ClusterIpDeployment", metadata=k8s.ObjectMeta(name=app_name), spec=k8s.DeploymentSpec( replicas=2, selector=k8s.LabelSelector(match_labels=label), template=k8s.PodTemplateSpec( metadata=k8s.ObjectMeta(labels=label, name=app_name), spec=k8s.PodSpec( containers=[k8s.Container( name=app_name, image="katacoda/docker-http-server:latest", ports=[k8s.ContainerPort(container_port=80)] )] ) ) ) ) k8s.KubeService( self, 'ClusterIpService', metadata=k8s.ObjectMeta(name=app_name), spec=k8s.ServiceSpec( type='NodePort', ports=[k8s.ServicePort(port=80, node_port=30080, name="80")], selector=label ) ) k8s.KubeService( self, 'ClusterIpTargetPortService', metadata=k8s.ObjectMeta(name=f"{app_name}-targetport"), spec=k8s.ServiceSpec( ports=[k8s.ServicePort(port=8080, target_port=k8s.IntOrString.from_number(80), name="8080")], selector=label ) ) 
Enter fullscreen mode Exit fullscreen mode
 #!/usr/bin/env python from constructs import Construct from cdk8s import App, Chart from clusterip import ClusterIp class MyChart(Chart): def __init__(self, scope: Construct, id: str): super().__init__(scope, id) ClusterIp(self, 'clusterip') app = App() MyChart(app, "clusterip") app.synth() 
Enter fullscreen mode Exit fullscreen mode

2. Run

 

⚑ $ cdk8s synth
dist/clusterip.k8s.yaml

⚑ $ cat dist/clusterip.k8s.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: wappip
spec:
replicas: 2
selector:
matchLabels:
app: wappip
template:
metadata:
labels:
app: wappip
name: wappip
spec:
containers:
- image: katacoda/docker-http-server:latest
name: wappip
ports:

- containerPort: 80

apiVersion: v1
kind: Service
metadata:
name: wappip
spec:
ports:
- name: "80"
nodePort: 30080
port: 80
selector:
app: wappip

type: NodePort

apiVersion: v1
kind: Service
metadata:
name: wappip-targetport
spec:
ports:
- name: "8080"
port: 8080
targetPort: 80
selector:
app: wappip

Enter fullscreen mode Exit fullscreen mode




πŸš€ Create deployment app with readiness and liveness probes


 

from constructs import Construct
from imports import k8s
import re

class ReadinessLivenesProbes(Construct):
def init(self, scope: Construct, id: str):
super().init(scope, id)

 for _name in ['frontend', 'bad-frontend']: if re.search('bad', _name): image_tag = 'unhealthy' toleration = k8s.Toleration() affinity = k8s.Affinity() else: image_tag = 'health' toleration = k8s.Toleration( effect='NoSchedule', key='fe', operator='Equal', value='healthy' ) affinity = k8s.Affinity( node_affinity=k8s.NodeAffinity( required_during_scheduling_ignored_during_execution=k8s.NodeSelector( node_selector_terms=[ k8s.NodeSelectorTerm( match_expressions=[ k8s.NodeSelectorRequirement( key='kubernetes.io/hostname', operator='In', values=['kube1'] ) ] ) ] ) ) ) label = {'app': _name} k8s.KubeDeployment( self, f"FrontendHealthy{_name}", metadata=k8s.ObjectMeta(labels=label, name=_name), spec=k8s.DeploymentSpec( replicas=1, selector=k8s.LabelSelector(match_labels=label), template=k8s.PodTemplateSpec( metadata=k8s.ObjectMeta(labels=label), spec=k8s.PodSpec( containers=[ k8s.Container( name=_name, image=f"katacoda/docker-http-server:{image_tag}", ports=[k8s.ContainerPort(container_port=80)], resources=k8s.ResourceRequirements( limits={"memory": k8s.Quantity.from_string('3000Mi'), "cpu": k8s.Quantity.from_string('1000m')}, requests={"memory": k8s.Quantity.from_string('300Mi'), "cpu": k8s.Quantity.from_string('200m')} ), readiness_probe=k8s.Probe( http_get=k8s.HttpGetAction(port=k8s.IntOrString.from_number(80), path='/'), initial_delay_seconds=1, timeout_seconds=1 ), liveness_probe=k8s.Probe( http_get=k8s.HttpGetAction(port=k8s.IntOrString.from_number(80), path='/'), initial_delay_seconds=1, failure_threshold=3, timeout_seconds=1 ) ) ], tolerations=[toleration], affinity=affinity ) ) ) ) 
Enter fullscreen mode Exit fullscreen mode
Enter fullscreen mode Exit fullscreen mode




πŸš€ Create statefulset with persistent volume


 

from constructs import Construct, Node
from imports import k8s

class StateFulSet(Construct):
def init(self, scope: Construct, id: str):
super().init(scope, id)

 msql_name = 'msql' msql_label = {'ss': msql_name} k8s.KubeStatefulSet( self, "MsqlStatefulSet", metadata=k8s.ObjectMeta(name=msql_name), spec=k8s.StatefulSetSpec( replicas=1, selector=k8s.LabelSelector(match_labels=msql_label), service_name=msql_name, template=k8s.PodTemplateSpec( metadata=k8s.ObjectMeta(labels=msql_label, name=msql_name), spec=k8s.PodSpec( containers=[k8s.Container( name=msql_name, image="openshift/mysql-55-centos7", ports=[k8s.ContainerPort(container_port=3306)], env=[ k8s.EnvVar(name='MYSQL_ROOT_PASSWORD', value='yourpassword'), k8s.EnvVar(name='MYSQL_USER', value='wp_user'), k8s.EnvVar(name='MYSQL_PASSWORD', value='wp_pass'), k8s.EnvVar(name='MYSQL_DATABASE', value='wp_db'), ], volume_mounts=[k8s.VolumeMount(mount_path='/var/lib/mysql/data', name='mysql-persistent-storage')] )] ) ), volume_claim_templates=[ k8s.KubePersistentVolumeClaimProps( metadata=k8s.ObjectMeta(name="mysql-persistent-storage"), spec=k8s.PersistentVolumeClaimSpec( access_modes=["ReadWriteOnce"], resources=k8s.ResourceRequirements(requests={"storage": k8s.Quantity.from_string("2Gi")}) ) ) ] ) ) k8s.KubePersistentVolume( self, "STSPersistentVolume1", metadata=k8s.ObjectMeta(name='nfs-0001'), spec=k8s.PersistentVolumeSpec( access_modes=['ReadWriteOnce', 'ReadWriteMany'], capacity={'storage': k8s.Quantity.from_string('2Gi')}, nfs=k8s.NfsVolumeSource(path='/exports/data-0001', server='192.168.121.210'), persistent_volume_reclaim_policy='Retain' ) ) k8s.KubePersistentVolume( self, "STSPersistentVolume2", metadata=k8s.ObjectMeta(name='nfs-0002'), spec=k8s.PersistentVolumeSpec( access_modes=['ReadWriteOnce', 'ReadWriteMany'], capacity={'storage': k8s.Quantity.from_string('5Gi')}, nfs=k8s.NfsVolumeSource(path='/exports/data-0002', server='192.168.121.210'), persistent_volume_reclaim_policy='Retain' ) ) 
Enter fullscreen mode Exit fullscreen mode
Enter fullscreen mode Exit fullscreen mode




πŸš€ Create secret attach to deployment


 

from constructs import Construct, Node
from imports import k8s

class Secrete(Construct):
def init(self, scope: Construct, id: str):
super().init(scope, id)

 app_name = 'myscecret' label = {'app': app_name} k8s.KubeDeployment( self, "SecreteDeployment", metadata=k8s.ObjectMeta(name=app_name), spec=k8s.DeploymentSpec( replicas=1, selector=k8s.LabelSelector(match_labels=label), template=k8s.PodTemplateSpec( metadata=k8s.ObjectMeta(labels=label, name=app_name), spec=k8s.PodSpec( containers=[ k8s.Container( name=app_name, image="alpine:latest", ports=[k8s.ContainerPort(container_port=80)], command=['sleep', '9999'], env=[ k8s.EnvVar( name='SECRET_USERNAME', value_from=k8s.EnvVarSource( secret_key_ref=k8s.SecretKeySelector(key='username', name='test-secret') ) ), k8s.EnvVar( name='SECRET_PASSWORD', value_from=k8s.EnvVarSource( secret_key_ref=k8s.SecretKeySelector(key='password', name='test-secret') ) ) ] ) ] ) ) ) ) k8s.KubeSecret( self, 'Secrete', metadata=k8s.ObjectMeta(name='test-secret'), type='Opaque', data={"username": "YWRtaW4=", "password": "YTYyZmpiZDM3OTQyZGNz"} ) 
Enter fullscreen mode Exit fullscreen mode
Enter fullscreen mode Exit fullscreen mode




πŸš€ Create cronjob with persistent volume claim


 

from constructs import Construct, Node
from imports import k8s

class DocCronjob(Construct):
def init(self, scope: Construct, id: str):
super().init(scope, id)

 app_name = 'doc-clean' label = {'app': app_name} k8s.KubeCronJobV1Beta1( self, "DocCronjob", metadata=k8s.ObjectMeta(name=app_name, labels=label), spec=k8s.CronJobSpec( job_template=k8s.JobTemplateSpec( metadata=k8s.ObjectMeta(name=app_name), spec=k8s.JobSpec( template=k8s.PodTemplateSpec( metadata=k8s.ObjectMeta(name=app_name), spec=k8s.PodSpec( containers=[ k8s.Container( name=app_name, image="busybox", volume_mounts=[k8s.VolumeMount(mount_path='/opt/Documents', name='efs')], args=['/bin/sh', '-c', 'find /opt/Documents/Viewer/cache -type d -mtime +6 -exec rm -r {} \;; find /opt/Documents/Viewer -type f -mtime +6 -exec rm {} \;' ] ), ], restart_policy='OnFailure', volumes=[ k8s.Volume( name='efs', persistent_volume_claim=k8s.PersistentVolumeClaimVolumeSource( claim_name='efs-pvc' ) ) ] ) ) ) ), schedule='0 1 * * SAT' ) ) 
Enter fullscreen mode Exit fullscreen mode
Enter fullscreen mode Exit fullscreen mode




πŸš€ Conclusion

cdk8s is just being in beta version, you should consider to use it for production until it reaches a certain level of maturity (probably the first major version)


🌠 Blog · Github · stackoverflow · Linkedin · Group · Page · Twitter 🌠

Top comments (0)