johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package cloudprovider

import (
	"context"
	"errors"
	"fmt"

	corev1 "k8s.io/api/core/v1"
)

type NodeWriter interface {
	PatchProviderID(ctx context.Context, nodeName, providerID string) error

	SetLabels(ctx context.Context, nodeName string, set map[string]string, remove []string) error

	DeleteNode(ctx context.Context, nodeName string) error
}

type Options struct {
	Cluster string

	Instances Instances

	Nodes NodeWriter

	Log func(event string, fields map[string]any)
}

// ponytail: this is a hand-rolled subset of the Kubernetes cloud-provider node contracts -- node
// init (providerID and labels) and node lifecycle (delete on gone or draining) only. The ceiling is
// that it is not a cloudprovider.Interface, so it can serve no zone, route or load balancer. The
// upgrade path is k8s.io/cloud-provider/app with a cloudprovider.Interface once a load balancer or a
// zone concept exists here.
type Reconciler struct {
	cluster   string
	instances Instances
	nodes     NodeWriter
	log       func(event string, fields map[string]any)
}

func NewReconciler(o Options) (*Reconciler, error) {
	if o.Cluster == "" {
		return nil, errors.New(
			"cloudprovider: the controller needs the name of the cluster whose Nodes it owns; an " +
				"empty one would stamp provider IDs and labels that name no cluster")
	}
	if o.Instances == nil {
		return nil, errors.New(
			"cloudprovider: the controller has no instance source, so every Node would read as " +
				"unknown and be deleted")
	}
	if o.Nodes == nil {
		return nil, errors.New(
			"cloudprovider: the controller has no Node writer, so every decision would be made " +
				"and dropped")
	}
	log := o.Log
	if log == nil {
		log = func(string, map[string]any) {}
	}
	return &Reconciler{
		cluster:   o.Cluster,
		instances: o.Instances,
		nodes:     o.Nodes,
		log:       log,
	}, nil
}

func (r *Reconciler) Cluster() string {
	return r.cluster
}

func (r *Reconciler) Reconcile(ctx context.Context, nodeName string, node corev1.Node, present bool) error {
	if nodeName == "" {
		return errors.New("cloudprovider: reconcile was given an empty node name")
	}
	if !present {
		r.log("cloud controller manager: node gone from the cache", map[string]any{
			"cluster": r.cluster,
			"node":    nodeName,
		})
		return nil
	}

	inst, known, err := r.instances.Get(ctx, nodeName)
	if err != nil {
		return fmt.Errorf(
			"cloudprovider: the instance source for %s could not be read: %w. An unreadable "+
				"source is NOT an absent instance, and acting on it would delete the Node of a "+
				"guest that is still running its work, so nothing is written and the Node is "+
				"requeued", nodeName, err)
	}
	if known && inst.Cluster == "" {
		return fmt.Errorf(
			"cloudprovider: the instance record for %s names no cluster. A record with no "+
				"cluster is MALFORMED, not absent: it yields no provider ID, and stamping the "+
				"Node anyway would leave whatever identity it carries -- one from a previous "+
				"life -- in place while its cluster label was overwritten. Nothing is written "+
				"and the Node is requeued", nodeName)
	}
	if known && inst.NodeName == "" {
		return fmt.Errorf(
			"cloudprovider: the instance record for %s names no node. A record with no node is "+
				"MALFORMED, not absent: it yields no provider ID, and nothing may be written "+
				"against it, so the Node is requeued", nodeName)
	}

	patch, changed := Desired(node, inst, known)
	if !changed {
		return nil
	}

	if patch.Delete {
		if err := r.nodes.DeleteNode(ctx, nodeName); err != nil {
			return fmt.Errorf("cloudprovider: deleting the Node %s because %s: %w",
				nodeName, patch.Reason, err)
		}
		r.log("cloud controller manager: node deleted", map[string]any{
			"phase":   "Reconciled",
			"cluster": r.cluster,
			"node":    nodeName,
			"reason":  patch.Reason,
		})
		return nil
	}

	if patch.ProviderID != "" && node.Spec.ProviderID != patch.ProviderID {
		if err := r.nodes.PatchProviderID(ctx, nodeName, patch.ProviderID); err != nil {
			return fmt.Errorf("cloudprovider: stamping %s on the Node %s: %w",
				patch.ProviderID, nodeName, err)
		}
	}
	if len(patch.SetLabels)+len(patch.RemoveLabels) > 0 {
		if err := r.nodes.SetLabels(ctx, nodeName, patch.SetLabels, patch.RemoveLabels); err != nil {
			return fmt.Errorf("cloudprovider: labelling the Node %s: %w", nodeName, err)
		}
	}
	r.log("cloud controller manager: node stamped", map[string]any{
		"phase":      "Reconciled",
		"cluster":    r.cluster,
		"node":       nodeName,
		"providerID": patch.ProviderID,
		"reason":     patch.Reason,
	})
	return nil
}