cozystack/internal/fluxinstall/install.go

/*
Copyright 2025 The Cozystack Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package fluxinstall

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	"github.com/cozystack/cozystack/internal/manifestutil"

	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"
)

// Install installs Flux components using embedded manifests.
// It extracts the manifests and applies them to the cluster.
// The namespace is automatically determined from the Namespace object in the manifests.
func Install(ctx context.Context, k8sClient client.Client, writeEmbeddedManifests func(string) error) error {
	logger := log.FromContext(ctx)

	// Create temporary directory for manifests
	tmpDir, err := os.MkdirTemp("", "flux-install-*")
	if err != nil {
		return fmt.Errorf("failed to create temp directory: %w", err)
	}
	defer os.RemoveAll(tmpDir)

	// Extract embedded manifests (generated by cozypkg)
	manifestsDir := filepath.Join(tmpDir, "manifests")
	if err := os.MkdirAll(manifestsDir, 0755); err != nil {
		return fmt.Errorf("failed to create manifests directory: %w", err)
	}

	if err := writeEmbeddedManifests(manifestsDir); err != nil {
		return fmt.Errorf("failed to extract embedded manifests: %w", err)
	}

	// Find all YAML manifest files
	entries, err := os.ReadDir(manifestsDir)
	if err != nil {
		return fmt.Errorf("failed to read manifests directory: %w", err)
	}

	var manifestFiles []string
	for _, entry := range entries {
		if strings.HasSuffix(entry.Name(), ".yaml") {
			manifestFiles = append(manifestFiles, filepath.Join(manifestsDir, entry.Name()))
		}
	}

	if len(manifestFiles) == 0 {
		return fmt.Errorf("no YAML manifest files found in directory")
	}

	// Parse all manifest files
	var objects []*unstructured.Unstructured
	for _, manifestPath := range manifestFiles {
		objs, err := manifestutil.ParseManifestFile(manifestPath)
		if err != nil {
			return fmt.Errorf("failed to parse manifests from %s: %w", manifestPath, err)
		}
		objects = append(objects, objs...)
	}

	if len(objects) == 0 {
		return fmt.Errorf("no objects found in manifests")
	}

	// Inject KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT if set in operator environment
	if err := injectKubernetesServiceEnv(objects); err != nil {
		logger.Info("Failed to inject KUBERNETES_SERVICE_* env vars, continuing anyway", "error", err)
	}

	// Extract namespace from Namespace object in manifests
	namespace, err := extractNamespace(objects)
	if err != nil {
		return fmt.Errorf("failed to extract namespace from manifests: %w", err)
	}

	logger.Info("Installing Flux components", "namespace", namespace)

	// Apply manifests using server-side apply
	logger.Info("Applying Flux manifests", "count", len(objects), "files", len(manifestFiles), "namespace", namespace)
	if err := applyManifests(ctx, k8sClient, objects); err != nil {
		return fmt.Errorf("failed to apply manifests: %w", err)
	}

	logger.Info("Flux installation completed successfully")
	return nil
}

// applyManifests applies Kubernetes objects using server-side apply.
func applyManifests(ctx context.Context, k8sClient client.Client, objects []*unstructured.Unstructured) error {
	logger := log.FromContext(ctx)

	// Separate CRDs and namespaces from other resources
	var stageOne []*unstructured.Unstructured // CRDs and Namespaces
	var stageTwo []*unstructured.Unstructured // Everything else

	for _, obj := range objects {
		if isClusterDefinition(obj) {
			stageOne = append(stageOne, obj)
		} else {
			stageTwo = append(stageTwo, obj)
		}
	}

	// Apply stage one (CRDs and Namespaces) first
	if len(stageOne) > 0 {
		logger.Info("Applying cluster definitions", "count", len(stageOne))
		if err := applyObjects(ctx, k8sClient, stageOne); err != nil {
			return fmt.Errorf("failed to apply cluster definitions: %w", err)
		}

		// Wait for CRDs to be established before applying dependent resources
		crdNames := manifestutil.CollectCRDNames(stageOne)
		if err := manifestutil.WaitForCRDsEstablished(ctx, k8sClient, crdNames); err != nil {
			return fmt.Errorf("CRDs not established after apply: %w", err)
		}
	}

	// Apply stage two (everything else)
	if len(stageTwo) > 0 {
		logger.Info("Applying resources", "count", len(stageTwo))
		if err := applyObjects(ctx, k8sClient, stageTwo); err != nil {
			return fmt.Errorf("failed to apply resources: %w", err)
		}
	}

	return nil
}

// applyObjects applies a list of objects using server-side apply.
func applyObjects(ctx context.Context, k8sClient client.Client, objects []*unstructured.Unstructured) error {
	for _, obj := range objects {
		// Use server-side apply with force ownership and field manager
		// FieldManager is required for apply patch operations
		patchOptions := &client.PatchOptions{
			FieldManager: "cozystack-operator",
			Force:        func() *bool { b := true; return &b }(),
		}

		if err := k8sClient.Patch(ctx, obj, client.Apply, patchOptions); err != nil {
			return fmt.Errorf("failed to apply object %s/%s: %w", obj.GetKind(), obj.GetName(), err)
		}
	}
	return nil
}

// extractNamespace extracts the namespace name from the Namespace object in the manifests.
func extractNamespace(objects []*unstructured.Unstructured) (string, error) {
	for _, obj := range objects {
		if obj.GetKind() == "Namespace" {
			namespace := obj.GetName()
			if namespace == "" {
				return "", fmt.Errorf("Namespace object has no name")
			}
			return namespace, nil
		}
	}
	return "", fmt.Errorf("no Namespace object found in manifests")
}

// isClusterDefinition checks if an object is a CRD or Namespace.
func isClusterDefinition(obj *unstructured.Unstructured) bool {
	kind := obj.GetKind()
	return kind == "CustomResourceDefinition" || kind == "Namespace"
}

// injectKubernetesServiceEnv injects KUBERNETES_SERVICE_HOST and KUBERNETES_SERVICE_PORT
// environment variables into all containers of Deployment, StatefulSet, and DaemonSet objects
// if these variables are set in the operator's environment.
// Errors are logged but do not stop processing of other objects.
func injectKubernetesServiceEnv(objects []*unstructured.Unstructured) error {
	kubernetesHost := os.Getenv("KUBERNETES_SERVICE_HOST")
	kubernetesPort := os.Getenv("KUBERNETES_SERVICE_PORT")

	// If neither variable is set, nothing to do
	if kubernetesHost == "" && kubernetesPort == "" {
		return nil
	}

	var firstErr error
	for _, obj := range objects {
		kind := obj.GetKind()
		if kind != "Deployment" && kind != "StatefulSet" && kind != "DaemonSet" {
			continue
		}

		// Navigate to spec.template.spec
		spec, found, err := unstructured.NestedMap(obj.Object, "spec", "template", "spec")
		if !found {
			continue
		}

		// Skip pods that don't use hostNetwork - they should use normal Kubernetes DNS
		hostNetwork, _, _ := unstructured.NestedBool(spec, "hostNetwork")
		if !hostNetwork {
			continue
		}
		if err != nil {
			if firstErr == nil {
				firstErr = fmt.Errorf("failed to get spec for %s/%s: %w", kind, obj.GetName(), err)
			}
			continue
		}

		// Update containers
		containers, found, err := unstructured.NestedSlice(spec, "containers")
		if err != nil {
			if firstErr == nil {
				firstErr = fmt.Errorf("failed to get containers for %s/%s: %w", kind, obj.GetName(), err)
			}
			continue
		}
		if found {
			containers = updateContainersEnv(containers, kubernetesHost, kubernetesPort)
			if err := unstructured.SetNestedSlice(spec, containers, "containers"); err != nil {
				if firstErr == nil {
					firstErr = fmt.Errorf("failed to set containers for %s/%s: %w", kind, obj.GetName(), err)
				}
				continue
			}
		}

		// Update initContainers
		initContainers, found, err := unstructured.NestedSlice(spec, "initContainers")
		if err != nil {
			if firstErr == nil {
				firstErr = fmt.Errorf("failed to get initContainers for %s/%s: %w", kind, obj.GetName(), err)
			}
			continue
		}
		if found {
			initContainers = updateContainersEnv(initContainers, kubernetesHost, kubernetesPort)
			if err := unstructured.SetNestedSlice(spec, initContainers, "initContainers"); err != nil {
				if firstErr == nil {
					firstErr = fmt.Errorf("failed to set initContainers for %s/%s: %w", kind, obj.GetName(), err)
				}
				continue
			}
		}

		// Update spec in the object
		if err := unstructured.SetNestedMap(obj.Object, spec, "spec", "template", "spec"); err != nil {
			if firstErr == nil {
				firstErr = fmt.Errorf("failed to update spec for %s/%s: %w", kind, obj.GetName(), err)
			}
			continue
		}
	}

	return firstErr
}

// updateContainersEnv updates environment variables for a slice of containers.
func updateContainersEnv(containers []any, kubernetesHost, kubernetesPort string) []any {
	for i, container := range containers {
		containerMap, ok := container.(map[string]any)
		if !ok {
			continue
		}

		env, found, err := unstructured.NestedSlice(containerMap, "env")
		if err != nil {
			continue
		}

		if !found {
			env = []any{}
		}

		// Update or add KUBERNETES_SERVICE_HOST
		if kubernetesHost != "" {
			env = setEnvVar(env, "KUBERNETES_SERVICE_HOST", kubernetesHost)
		}

		// Update or add KUBERNETES_SERVICE_PORT
		if kubernetesPort != "" {
			env = setEnvVar(env, "KUBERNETES_SERVICE_PORT", kubernetesPort)
		}

		// Update the container's env
		if err := unstructured.SetNestedSlice(containerMap, env, "env"); err != nil {
			continue
		}

		// Update the container in the slice
		containers[i] = containerMap
	}

	return containers
}

// setEnvVar updates or adds an environment variable in the env slice.
func setEnvVar(env []any, name, value string) []any {
	// Check if variable already exists
	for i, envVar := range env {
		envVarMap, ok := envVar.(map[string]any)
		if !ok {
			continue
		}

		if envVarMap["name"] == name {
			// Update existing variable
			envVarMap["value"] = value
			env[i] = envVarMap
			return env
		}
	}

	// Add new variable
	env = append(env, map[string]any{
		"name":  name,
		"value": value,
	})

	return env
}