DEV Community

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

Posted on • Updated 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)