Custom Resource Definition by hand

This page explains how to define manually the data model and client module for custom resources, without relying on the zio-k8s-crd code generator plugin.

We are going to use the Crontab example from the Kubernetes documentation.

First we define the model for our Crontab resource. This should be a case class and it can use the zio-k8s specific Optional type for optional fields instead of standard Option to reduce boilerplate when defining resources from code. It should have a standard metadata field, and optionally a status field:

case class Crontab(
  metadata: Optional[ObjectMeta],
  status: Optional[CrontabStatus],
  spec: String,
  image: String,
  replicas: Int
)

case class CrontabStatus(replicas: Int, labelSelector: String)

There is a couple of type classes that has to be implemented for these data structures. First of all, it has to support being encoded to and decoded from JSON. This can be done by providing a JSON codec:

implicit val crontabCodec: Codec[Crontab] = deriveCodec
// crontabCodec: Codec[Crontab] = io.circe.generic.codec.DerivedAsObjectCodec$$anon$1@eb5feaf
implicit val crontabStatusCodec: Codec[CrontabStatus] = deriveCodec
// crontabStatusCodec: Codec[CrontabStatus] = io.circe.generic.codec.DerivedAsObjectCodec$$anon$1@74ac4c90

Note that in practice you should put each implicit in the related type's companion object.

In order to be able to use the generic functions in zio-k8s, you have to provide an instance of the K8sObject and K8sObjectStatus type classes, defining how to get and manipulate the metadata and status fields of the custom resource:

implicit val k8sObject: K8sObject[Crontab] =
  new K8sObject[Crontab] {
    override def metadata(obj: Crontab): Optional[ObjectMeta] = obj.metadata
    override def mapMetadata(f: ObjectMeta => ObjectMeta)(r: Crontab): Crontab =
      r.copy(metadata = r.metadata.map(f))
  }
// k8sObject: K8sObject[Crontab] = repl.MdocSession$MdocApp$$anon$3@7d0b56d9

  implicit val k8sObjectStatus: K8sObjectStatus[Crontab, CrontabStatus] =
    new K8sObjectStatus[Crontab, CrontabStatus] {
      override def status(obj: Crontab): Optional[CrontabStatus] = obj.status
      override def mapStatus(f: CrontabStatus => CrontabStatus)(obj: Crontab): Crontab =
        obj.copy(status = obj.status.map(f))
    }
// k8sObjectStatus: K8sObjectStatus[Crontab, CrontabStatus] = repl.MdocSession$MdocApp$$anon$4@2dcade43

Finally we need to provide some metadata about the resource type with the ResourceMetadata type class:

implicit val metadata: ResourceMetadata[Crontab] = new ResourceMetadata[Crontab] {
  override def kind: String = "CronTab"
  override def apiVersion: String = "apiextensions.k8s.io/v1"
  override def resourceType: model.K8sResourceType = K8sResourceType("crontabs", "apiextensions.k8s.io", "v1")
}
// metadata: ResourceMetadata[Crontab] = repl.MdocSession$MdocApp$$anon$5@7a056b51

This is all we need for the model of our custom resources. The next step is creating a ZIO module for working with these resources through the Kubernetes API.

The following code snippet shows the recommended way to define these modules (crontabs can usually defined as a package object instead) :

object crontabs {
  type Crontabs = Crontabs.Service

  object Crontabs {
    // Generic representation (optional) 
    type Generic = NamespacedResource[Crontab] with NamespacedResourceStatus[CrontabStatus, Crontab]

    trait Service
      extends NamespacedResource[Crontab] with NamespacedResourceStatus[CrontabStatus, Crontab]

    final class Live(
      override val asGenericResource: ResourceClient[Crontab, Status],
      override val asGenericResourceStatus: ResourceStatusClient[CrontabStatus, Crontab]
    ) extends Service

    val live
      : ZLayer[K8sCluster with K8sBackend, Nothing, Crontabs] = 
        ZLayer {
          for {
            backend <- ZIO.service[K8sBackend]
            cluster <- ZIO.service[K8sCluster]
            client = new ResourceClient[Crontab, Status](metadata.resourceType, cluster, backend)
            statusClient = new ResourceStatusClient[CrontabStatus, Crontab](
              metadata.resourceType,
              cluster,
              backend
            )
          } yield new Live(client, statusClient)
        }
  }
}

In addition to this you can add accessor functions to the crontabs package in the style of

def get(name: String, namespace: K8sNamespace): ZIO[Crontabs, K8sFailure, Crontab] =
  ZIO.serviceWithZIO(_.get(name, namespace))