This post explains about a kubernetes controller that I've created.
I've had reviewed several guides for kubernetes controller creation, such as
Extending Kubernetes,
Kubernetes Custom Controller, and
this. I've found myself overwhelmed by the complexity of such a simple requirement. This post is the present the KISS gist of my blog. When you need to perform a task, ask yourself, does it really need to be so complex? Do I need all this monster for a simple task?
Many controllers are based on kubernetes Custom Resource Definition (aka CRD).
CRDs are indeed nice, but make the implementation much more complex, as you will have to use code generator tools.
If this is an internal solution, do you really need CRD? Why not use a simple ConfigMap?
In the solution presented below, I am using a ConfigMap that configures the number of StatefulSets to create. The ConfigMap also contains the template that will be used for the StatefulSet creation.
An example of the ConfigMap is:
apiVersion: v1
kind: ConfigMap
metadata:
labels:
resource-type: "primary"
name: primary-config
namespace: default
data:
general.yaml: |-
count: 2
statefulSet.yaml: "apiVersion: apps/v1\nkind: StatefulSet\nmetadata:\n name: my-statefulset-___sequence___\n
\ labels:\n resource-type: \"secondary\" \n app.kubernetes.io/instance
: bouncer\n app.kubernetes.io/name : bouncer\nspec:\n
...
The controller main code is the merge logic, which fetch the primary ConfigMap, and creates/delete StatefulSet accordingly. This is done by the following logic:
- Get the primary ConfigMap
- Load the existing StatefulSets
- Loop until the required count in the primary ConfigMap
- Create the StatefulSet if it does not exist
- Delete all StatefulSets that were not encountered in the loop above
package mypackage
import (
"fmt"
"gopkg.in/yaml.v2"
apps "k8s.io/api/apps/v1"
core "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes/scheme"
"strconv"
"strings"
"sync"
)
type Merger struct {
updates chan int
api *K8sApi
primaryConfig *core.ConfigMap
currentEntities []apps.StatefulSet
requiredEntities []apps.StatefulSet
}
func (merger *Merger) Init() {
merger.updates = make(chan int, 100)
go func() {
for {
select {
case update := <-merger.updates:
for len(merger.updates) > 0 {
<-merger.updates
}
merger.merge()
}
}
}()
}
func (merger *Merger) NotifyUpdate() {
merger.updates <- MessageUpdate
}
func (merger *Merger) merge() {
configMaps := merger.api.configMap.FetchByLabels("resource-type=primary")
merger.primaryConfig = &configMaps[0]
merger.currentEntities = merger.api.statefulSet.FetchByLabels("resource-type=secondary")
merger.requiredEntities = []apps.StatefulSet{}
chains := merger.getRequiredCount()
for i := 0; i < chains; i++ {
merger.mergeSequence(i)
}
merger.deleteNonUsedStatefulSets()
}
func (merger *Merger) deleteNonUsedStatefulSets() {
usedNames := map[string]bool{}
for _, s := range merger.requiredEntities {
usedNames[s.Name] = true
}
for _, s := range merger.currentEntities {
if !usedNames[s.Name] {
merger.api.statefulSet.Delete(s.Name)
}
}
}
func (merger *Merger) mergeSequence(sequence int) {
statefulSetText := merger.getStatefulSetUpdatedText(sequence, name)
decode := scheme.Codecs.UniversalDeserializer().Decode
obj, _, err := decode([]byte(statefulSetText), nil, nil)
if err != nil {
Fatal("decode statefulSetText failed, %v", err)
}
statefulSet, ok := obj.(*apps.StatefulSet)
if !ok {
Fatal("cast statefulSet failed")
}
merger.requiredEntities = append(merger.requiredEntities, *statefulSet)
if merger.statefulSetExists(statefulSet.Name) {
return
}
merger.api.statefulSet.Create(statefulSet)
}
func (merger *Merger) getStatefulSetUpdatedText(sequence int, name string) string {
statefulSet := merger.primaryConfig.Data["statefulSet.yaml"]
statefulSet = merger.templateReplace(statefulSet, "sequence", strconv.Itoa(sequence))
statefulSet = merger.templateReplace(statefulSet, "name", name)
replacements := merger.getGeneralConfig()[name]
return statefulSet
}
func (merger *Merger) templateReplace(text string, from string, to string) string {
return strings.Replace(text, fmt.Sprintf("___%v___", from), to, -1)
}
func (merger *Merger) getRequiredCount() int {
generalConfig := merger.getGeneralConfig()
countStr := generalConfig["count"]
count, err := strconv.Atoi(fmt.Sprintf("%v", countStr))
if err != nil {
Fatal("parse of count %v failed %v", countStr, err)
}
return count
}
func (merger *Merger) getGeneralConfig() map[interface{}]interface{} {
general := merger.primaryConfig.Data["general.yaml"]
generalConfig := make(map[interface{}]interface{})
err := yaml.Unmarshal([]byte(general), &generalConfig)
if err != nil {
Fatal("error parsing configMap yaml %v", err)
}
return generalConfig
}
func (merger *Merger) statefulSetExists(name string) bool {
for _, statefulSet := range merger.currentEntities.statefulSets {
if statefulSet.Name == name {
return true
}
}
return false
}
All we need now is to activate the NotifyUpdate upon change.
For this purpose we use the kubernetes watch API. We should watch for both the ConfigMap, and for StatefulSet changes. Watch for the StatefuleSet for example is done by the following:
func (api *K8sStatefulSetApi) WatchByLabels(labels string, notifier *Notifier) {
listOptions := meta.ListOptions{
LabelSelector: labels,
}
watcher, err := client.AppsV1().StatefulSets("default").Watch(listOptions)
if err != nil {
Fatal("watch kubernetes statefulSet failed: %v", err)
}
ch := watcher.ResultChan()
go func() {
Verbose("watcher loop for statefulSet established")
for event := range ch {
statefulSet, ok := event.Object.(*apps.StatefulSet)
if !ok {
Fatal("unexpected event in watcher channel")
}
Verbose("got event for statefulSet %v", statefulSet.Name)
(*notifier)()
}
}()
}
Summary
In this post a simple kubernetes controller based on a ConfigMap was presented.
The primary resource is the configMap, marked by the label: "resource-type=primary"
The secondary resources are the StatefulSets, marked by the label: "resource-type=secondary"
This controller does not use CRD, hence it does not require any code generation tools, and allows KISS implementation.
