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 (
	"bufio"
	"bytes"
	"context"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"time"

	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/serializer/yaml"
	k8syaml "k8s.io/apimachinery/pkg/util/yaml"
	"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 the manifest file (should be fluxcd.yaml from cozypkg)
	manifestPath := filepath.Join(manifestsDir, "fluxcd.yaml")
	if _, err := os.Stat(manifestPath); err != nil {
		// Try to find any YAML file if fluxcd.yaml doesn't exist
		entries, err := os.ReadDir(manifestsDir)
		if err != nil {
			return fmt.Errorf("failed to read manifests directory: %w", err)
		}
		for _, entry := range entries {
			if strings.HasSuffix(entry.Name(), ".yaml") {
				manifestPath = filepath.Join(manifestsDir, entry.Name())
				break
			}
		}
	}

	// Parse and apply manifests
	objects, err := parseManifests(manifestPath)
	if err != nil {
		return fmt.Errorf("failed to parse manifests: %w", err)
	}

	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), "manifest", manifestPath, "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
}

// parseManifests parses YAML manifests into unstructured objects.
func parseManifests(manifestPath string) ([]*unstructured.Unstructured, error) {
	data, err := os.ReadFile(manifestPath)
	if err != nil {
		return nil, fmt.Errorf("failed to read manifest file: %w", err)
	}

	return readYAMLObjects(bytes.NewReader(data))
}

// readYAMLObjects parses multi-document YAML into unstructured objects.
func readYAMLObjects(reader io.Reader) ([]*unstructured.Unstructured, error) {
	var objects []*unstructured.Unstructured
	yamlReader := k8syaml.NewYAMLReader(bufio.NewReader(reader))

	for {
		doc, err := yamlReader.Read()
		if err != nil {
			if err == io.EOF {
				break
			}
			return nil, fmt.Errorf("failed to read YAML document: %w", err)
		}

		// Skip empty documents
		if len(bytes.TrimSpace(doc)) == 0 {
			continue
		}

		obj := &unstructured.Unstructured{}
		decoder := k8syaml.NewYAMLOrJSONDecoder(bytes.NewReader(doc), len(doc))
		if err := decoder.Decode(obj); err != nil {
			// Skip documents that can't be decoded (might be comments or empty)
			if err == io.EOF {
				continue
			}
			return nil, fmt.Errorf("failed to decode YAML document: %w", err)
		}

		// Skip empty objects (no kind)
		if obj.GetKind() == "" {
			continue
		}

		objects = append(objects, obj)
	}

	return objects, 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)
	decoder := yaml.NewDecodingSerializer(unstructured.UnstructuredJSONScheme)

	// 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, decoder, stageOne); err != nil {
			return fmt.Errorf("failed to apply cluster definitions: %w", err)
		}

		// Wait a bit for CRDs to be registered
		time.Sleep(2 * time.Second)
	}

	// Apply stage two (everything else)
	if len(stageTwo) > 0 {
		logger.Info("Applying resources", "count", len(stageTwo))
		if err := applyObjects(ctx, k8sClient, decoder, 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, decoder runtime.Decoder, 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.
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
	}

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

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

		// Update containers
		containers, found, err := unstructured.NestedSlice(spec, "containers")
		if found && err == nil {
			containers = updateContainersEnv(containers, kubernetesHost, kubernetesPort)
			if err := unstructured.SetNestedSlice(spec, containers, "containers"); err != nil {
				continue
			}
		}

		// Update initContainers
		initContainers, found, err := unstructured.NestedSlice(spec, "initContainers")
		if found && err == nil {
			initContainers = updateContainersEnv(initContainers, kubernetesHost, kubernetesPort)
			if err := unstructured.SetNestedSlice(spec, initContainers, "initContainers"); err != nil {
				continue
			}
		}

		// Update spec in the object
		if err := unstructured.SetNestedMap(obj.Object, spec, "spec", "template", "spec"); err != nil {
			continue
		}
	}

	return nil
}

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

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

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

		// 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 []interface{}, name, value string) []interface{} {
	// Check if variable already exists
	for i, envVar := range env {
		envVarMap, ok := envVar.(map[string]interface{})
		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]interface{}{
		"name":  name,
		"value": value,
	})

	return env
}