/* 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 }