Reconcile all kustomizations

[rkthtrifork]

Hello

My team and are are developing a kubernetes cluster and we have a local setup using kind which is based on the same approach as flux-local-dev. We often need to push new changes to the local registry and trigger a flux reconcile, and there doesnt seem to be a command in the flux cli that reconciles all kustomizations as soon as possible.

We started with a loop to just start reconciling each kustomization, but that ended up being the wrong order (in terms of dependencies, i.e. spec.dependsOn) and therefore very slow. We then used a topological sort based on the kustomization dependencies which was better. But it runs in bash and we did not want to go through the effort of making it multi threaded in bash so it was single threaded and therefore also not great.

Now we have a go program which runs flux reconcile on a kustomization as soon as it is ready and every flux reconcile is run in a separate thread so reconciliations arent blocked. Is there a solution for this in the flux cli or can we add the functionality? I think i have some time to make a PR soon and if i do have time i would love to. Heres the code we are currently using

package main

import (
	"encoding/json"
	"flag"
	"log"
	"os/exec"
	"sync"
)

type KustomizationJson struct {
	Metadata struct {
		Name      string `json:"name"`
		Namespace string `json:"namespace"`
	} `json:"metadata"`
	Spec struct {
		DependsOn []struct {
			Name string `json:"name"`
		} `json:"dependsOn"`
	} `json:"spec"`
}

type KustomizationJsonList struct {
	Items []KustomizationJson `json:"items"`
}

type FQN struct {
	Namespace string
	Name      string
}

type Kustomization struct {
	Name           string
	Namespace      string
	DependsOn      []*Kustomization
	IsDependedOnBy []*Kustomization
	InDegree       int
}

var (
	wg               sync.WaitGroup
	reconcileTimeout = flag.String("timeout", "10m0s", "Flux reconcile timeout")
	kubeContext      = flag.String("context", "", "Kubernetes context")
)

func main() {
	flag.Parse()

	// Get all kustomizations
	out, err := exec.Command("kubectl", "get", "kustomizations", "--context", *kubeContext, "-A", "-o", "json").Output()
	if err != nil {
		log.Fatalf("Error calling kubectl: %v", err)
	}

	var kustomizationsJson KustomizationJsonList
	if err := json.Unmarshal(out, &kustomizationsJson); err != nil {
		log.Fatalf("Failed to parse JSON: %v", err)
	}

	// Create kustomization objects
	kustomizations := make(map[FQN]*Kustomization)
	for _, k := range kustomizationsJson.Items {
		FQN := FQN{
			Namespace: k.Metadata.Namespace,
			Name:      k.Metadata.Name,
		}
		kustomizations[FQN] = &Kustomization{
			Name:           k.Metadata.Name,
			Namespace:      k.Metadata.Namespace,
			DependsOn:      []*Kustomization{},
			IsDependedOnBy: []*Kustomization{},
			InDegree:       len(k.Spec.DependsOn),
		}
	}

	// Populate dependencies
	for _, k := range kustomizationsJson.Items {
		dependsOn := []*Kustomization{}
		for _, dep := range k.Spec.DependsOn {
			FQN := FQN{
				Namespace: k.Metadata.Namespace,
				Name:      dep.Name,
			}
			dependsOn = append(dependsOn, kustomizations[FQN])
		}
		FQN := FQN{
			Namespace: k.Metadata.Namespace,
			Name:      k.Metadata.Name,
		}
		kustomizations[FQN].DependsOn = dependsOn
		kustomizations[FQN].InDegree = len(k.Spec.DependsOn)
	}

	for _, kustomization := range kustomizations {
		for _, dep := range kustomization.DependsOn {
			dep.IsDependedOnBy = append(dep.IsDependedOnBy, kustomization)
		}
	}

	// Start reconciling
	for _, kustomization := range kustomizations {
		if kustomization.InDegree == 0 {
			wg.Add(1)
			go reconcile(kustomization)
		}
	}

	wg.Wait()
}

func reconcile(k *Kustomization) {
	defer wg.Done()

	log.Printf("[START] Reconcile: %s/%s\n", k.Namespace, k.Name)

	// Reconcile kustomization
	cmd := exec.Command(
		"flux", "reconcile", "kustomization",
		k.Name,
		"--context", *kubeContext,
		"--namespace", k.Namespace,
		"--timeout", *reconcileTimeout,
	)

	out, err := cmd.CombinedOutput()
	if err != nil {
		log.Fatalf("[ERROR] flux reconcile failed for %s: %v\nOutput:\n%s\n", k.Name, err, out)
	}

	log.Printf("[DONE]  Reconcile: %s/%s\n", k.Namespace, k.Name)

	// Update in-degree of kustomizations that depend on this one
	for _, dep := range k.IsDependedOnBy {
		dep.InDegree--
		if dep.InDegree == 0 {
			wg.Add(1)
			go reconcile(dep)
		}
	}
}

[matheuscscp]

You can use the Receiver API to have a list of kustomizations to reconcile, then if you set spec.type to generic you can use curl locally to trigger a reconciliation of all those kustomizations, docs here:

https://fluxcd.io/flux/components/notification/receivers/#resources

To run the curl you will need either a kubectl port-forward or an Ingress or something allowing local traffic into the cluster.

[rkthtrifork]

Thanks for the suggestion. I have a couple of questions about how the Receiver API would handle dependencies:

1. Dependency Timing: When using the receiver to trigger reconciliation for a list of kustomizations, will Flux still respect the spec.dependsOn ordering? I'm concerned that if all kustomizations are notified at the same time, some may attempt reconciliation before their dependencies are ready, go into status dependency 'x' is not ready and wait the full spec.retryInterval before it attempts to reconcile again. This is what we are trying to avoid by only reconciling a kustomization as soon as all its dependencies are ready.

2. Comparison with flux reconcile kustomization: Could you clarify how using the receiver differs from simply running flux reconcile kustomization on each resource? Specifically, how does the receiver ensure that reconciliation doesn't proceed if a kustomization’s dependencies aren't yet satisfied?

[matheuscscp]

The Receiver API is exactly like the flux CLI, it just annotates the target objects for reconciliation. However, like the CLI it doesn't wait for the reconciliations. Also neither the CLI nor the Receiver API handle dependencies, that is entirely up to the respective Flux controller. Flux will always respect the dependsOn relationships because it's the controllers who implement that, CLI and Receiver don't have to worry about it. When the Receiver kicks in what you will see is the Kustomizations who depend on others with the message "dependency is not ready", then eventually converging and reconciling.

[rkthtrifork]

They do indeed eventually converge and reconcile, but sometimes the kustomize controller tries to reconcile a kustomization and then find out that the dependencies are not ready. This puts the kustomization in the state dependency 'x' is not ready and it will not be retried before the spec.interval time has passed. This creates a waiting period with a wors case time of spec.interval. We are avoiding this extra gap by tracking the dependencies through the Go program i posted. From my understanding, the Receiver API will not solve this

[matheuscscp]

will not be retried before the spec.interval time has passed

Not really, when blocked by dependencies the retry interval is another one, configured at the controller level with a flag of the binary, --requeue-dependency:

https://github.com/fluxcd/kustomize-controller/blob/dc0e5853c06df506de22ce1bb0b9303389d90942/internal/controller/kustomization_controller.go#L277

Docs: https://fluxcd.io/flux/components/kustomize/options/#flags

You can set this to, say, 5s.

[rkthtrifork]

I just tested --requeue-dependency and it seems to work. We have been unable to find that for half a year 😆

Would you be interested in having the feature for the Flux CLI tool as well? I think its a somewhat common situation that a user would like to manually trigger a reconciliation for all their kustomizations without having set the --requeue-dependency.

[matheuscscp]

Yeah, flux reconcile kustomization all could be nice.

[rkthtrifork]

I might look into that soon then, thank you very much for your replies, it has been a great help 😃