[cozystack-operator] Introduce Cozystack-operator core logic (#1741)

Signed-off-by: Andrei Kvapil <kvapss@gmail.com>  ## What this PR does ### Release note  ```release-note [cozystack-operator] Introduce Cozystack-operator core logic ```  ## Summary by CodeRabbit * **New Features** * Operator entrypoint with flag-driven configuration (metrics, probes, leader election, Flux pre-install, version, platform-source options). * Optional pre-start Flux installation and automatic provisioning of platform sources from OCI or Git/HTTP(S)/SSH. * **Chores** * Embedded Flux manifests packaged for simplified installation; build tooling updated to produce new manifest outputs. * **Behavior** * Pre-start Flux and platform-source installation failures are treated as fatal. <sub>✏️ Tip: You can customize this high-level summary in your review settings.</sub>
2025-12-30 11:48:13 +01:00 · 2025-12-30 11:48:13 +01:00 · 2d4e568efe
commit 2d4e568efe
parent 715509f57e db8a006565
7 changed files with 12845 additions and 8 deletions
--- a/cmd/cozystack-operator/main.go
+++ b/cmd/cozystack-operator/main.go
@ -0,0 +1,439 @@
+/*
+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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
+	// to ensure that exec-entrypoint and run can make use of them.
+	_ "k8s.io/client-go/plugin/pkg/client/auth"
+
+	cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
+	helmv2 "github.com/fluxcd/helm-controller/api/v2"
+	sourcev1 "github.com/fluxcd/source-controller/api/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+
+	"github.com/cozystack/cozystack/internal/fluxinstall"
+	// +kubebuilder:scaffold:imports
+)
+
+var (
+	scheme   = runtime.NewScheme()
+	setupLog = ctrl.Log.WithName("setup")
+)
+
+func init() {
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
+	utilruntime.Must(cozyv1alpha1.AddToScheme(scheme))
+	utilruntime.Must(helmv2.AddToScheme(scheme))
+	utilruntime.Must(sourcev1.AddToScheme(scheme))
+	// +kubebuilder:scaffold:scheme
+}
+
+func main() {
+	var metricsAddr string
+	var enableLeaderElection bool
+	var probeAddr string
+	var secureMetrics bool
+	var enableHTTP2 bool
+	var installFlux bool
+	var cozystackVersion string
+	var platformSourceURL string
+	var platformSourceName string
+	var platformSourceRef string
+
+	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
+	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
+	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
+		"Enable leader election for controller manager. "+
+			"Enabling this will ensure there is only one active controller manager.")
+	flag.BoolVar(&secureMetrics, "metrics-secure", false,
+		"If set the metrics endpoint is served securely")
+	flag.BoolVar(&enableHTTP2, "enable-http2", false,
+		"If set, HTTP/2 will be enabled for the metrics and webhook servers")
+	flag.BoolVar(&installFlux, "install-flux", false, "Install Flux components before starting reconcile loop")
+	flag.StringVar(&cozystackVersion, "cozystack-version", "unknown",
+		"Version of Cozystack")
+	flag.StringVar(&platformSourceURL, "platform-source-url", "", "Platform source URL (oci:// or https://). If specified, generates OCIRepository or GitRepository resource.")
+	flag.StringVar(&platformSourceName, "platform-source-name", "cozystack-packages", "Name for the generated platform source resource (default: cozystack-packages)")
+	flag.StringVar(&platformSourceRef, "platform-source-ref", "", "Reference specification as key=value pairs (e.g., 'branch=main' or 'digest=sha256:...,tag=v1.0'). For OCI: digest, semver, semverFilter, tag. For Git: branch, tag, semver, name, commit.")
+
+	opts := zap.Options{
+		Development: true,
+	}
+	opts.BindFlags(flag.CommandLine)
+	flag.Parse()
+
+	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
+
+	config := ctrl.GetConfigOrDie()
+
+	// Create a direct client (without cache) for pre-start operations
+	directClient, err := client.New(config, client.Options{Scheme: scheme})
+	if err != nil {
+		setupLog.Error(err, "unable to create direct client")
+		os.Exit(1)
+	}
+
+	// Start the controller manager
+	setupLog.Info("Starting controller manager")
+	mgr, err := ctrl.NewManager(config, ctrl.Options{
+		Scheme: scheme,
+		Metrics: metricsserver.Options{
+			BindAddress:   metricsAddr,
+			SecureServing: secureMetrics,
+		},
+		WebhookServer: webhook.NewServer(webhook.Options{
+			Port: 9443,
+		}),
+		HealthProbeBindAddress: probeAddr,
+		LeaderElection:         enableLeaderElection,
+		LeaderElectionID:       "cozystack-operator.cozystack.io",
+		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
+		// when the Manager ends. This requires the binary to immediately end when the
+		// Manager is stopped, otherwise, setting this significantly speeds up voluntary
+		// leader transitions as the new leader don't have to wait LeaseDuration time first.
+		//
+		// In the default scaffold provided, the program ends immediately after
+		// the manager stops, so would be fine to enable this option. However,
+		// if you are doing or is intended to do any operation such as perform cleanups
+		// after the manager stops then its usage might be unsafe.
+		// LeaderElectionReleaseOnCancel: true,
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to start manager")
+		os.Exit(1)
+	}
+
+	// Install Flux before starting reconcile loop
+	if installFlux {
+		setupLog.Info("Installing Flux components before starting reconcile loop")
+		installCtx, installCancel := context.WithTimeout(context.Background(), 5*time.Minute)
+		defer installCancel()
+
+		// Use direct client for pre-start operations (cache is not ready yet)
+		if err := fluxinstall.Install(installCtx, directClient, fluxinstall.WriteEmbeddedManifests); err != nil {
+			setupLog.Error(err, "failed to install Flux")
+			os.Exit(1)
+		}
+		setupLog.Info("Flux installation completed successfully")
+	}
+
+	// Generate and install platform source resource if specified
+	if platformSourceURL != "" {
+		setupLog.Info("Generating platform source resource", "url", platformSourceURL, "name", platformSourceName, "ref", platformSourceRef)
+		installCtx, installCancel := context.WithTimeout(context.Background(), 2*time.Minute)
+		defer installCancel()
+
+		// Use direct client for pre-start operations (cache is not ready yet)
+		if err := installPlatformSourceResource(installCtx, directClient, platformSourceURL, platformSourceName, platformSourceRef); err != nil {
+			setupLog.Error(err, "failed to install platform source resource")
+			os.Exit(1)
+		} else {
+			setupLog.Info("Platform source resource installation completed successfully")
+		}
+	}
+
+	// +kubebuilder:scaffold:builder
+
+	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+		setupLog.Error(err, "unable to set up health check")
+		os.Exit(1)
+	}
+	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+		setupLog.Error(err, "unable to set up ready check")
+		os.Exit(1)
+	}
+
+	setupLog.Info("Starting controller manager")
+	mgrCtx := ctrl.SetupSignalHandler()
+	if err := mgr.Start(mgrCtx); err != nil {
+		setupLog.Error(err, "problem running manager")
+		os.Exit(1)
+	}
+}
+
+// installPlatformSourceResource generates and installs a Flux source resource (OCIRepository or GitRepository)
+// based on the platform source URL
+func installPlatformSourceResource(ctx context.Context, k8sClient client.Client, sourceURL, resourceName, refSpec string) error {
+	logger := log.FromContext(ctx)
+
+	// Parse the source URL to determine type
+	sourceType, repoURL, err := parsePlatformSourceURL(sourceURL)
+	if err != nil {
+		return fmt.Errorf("failed to parse platform source URL: %w", err)
+	}
+
+	// Parse reference specification
+	refMap, err := parseRefSpec(refSpec)
+	if err != nil {
+		return fmt.Errorf("failed to parse reference specification: %w", err)
+	}
+
+	var obj client.Object
+	switch sourceType {
+	case "oci":
+		obj, err = generateOCIRepository(resourceName, repoURL, refMap)
+		if err != nil {
+			return fmt.Errorf("failed to generate OCIRepository: %w", err)
+		}
+	case "git":
+		obj, err = generateGitRepository(resourceName, repoURL, refMap)
+		if err != nil {
+			return fmt.Errorf("failed to generate GitRepository: %w", err)
+		}
+	default:
+		return fmt.Errorf("unsupported source type: %s (expected oci:// or https://)", sourceType)
+	}
+
+	// Apply the resource (create or update)
+	logger.Info("Applying platform source resource",
+		"apiVersion", obj.GetObjectKind().GroupVersionKind().GroupVersion().String(),
+		"kind", obj.GetObjectKind().GroupVersionKind().Kind,
+		"name", obj.GetName(),
+		"namespace", obj.GetNamespace(),
+	)
+
+	existing := obj.DeepCopyObject().(client.Object)
+	key := client.ObjectKeyFromObject(obj)
+
+	err = k8sClient.Get(ctx, key, existing)
+	if err != nil {
+		if client.IgnoreNotFound(err) == nil {
+			// Resource doesn't exist, create it
+			if err := k8sClient.Create(ctx, obj); err != nil {
+				return fmt.Errorf("failed to create resource %s/%s: %w", obj.GetObjectKind().GroupVersionKind().Kind, obj.GetName(), err)
+			}
+			logger.Info("Created platform source resource", "kind", obj.GetObjectKind().GroupVersionKind().Kind, "name", obj.GetName())
+		} else {
+			return fmt.Errorf("failed to check if resource exists: %w", err)
+		}
+	} else {
+		// Resource exists, update it
+		obj.SetResourceVersion(existing.GetResourceVersion())
+		if err := k8sClient.Update(ctx, obj); err != nil {
+			return fmt.Errorf("failed to update resource %s/%s: %w", obj.GetObjectKind().GroupVersionKind().Kind, obj.GetName(), err)
+		}
+		logger.Info("Updated platform source resource", "kind", obj.GetObjectKind().GroupVersionKind().Kind, "name", obj.GetName())
+	}
+
+	return nil
+}
+
+// parsePlatformSourceURL parses the source URL and returns the source type and repository URL.
+// Supports formats:
+//   - oci://registry.example.com/repo
+//   - https://github.com/user/repo
+//   - http://github.com/user/repo
+//   - ssh://git@github.com/user/repo
+func parsePlatformSourceURL(sourceURL string) (sourceType, repoURL string, err error) {
+	sourceURL = strings.TrimSpace(sourceURL)
+
+	if strings.HasPrefix(sourceURL, "oci://") {
+		return "oci", sourceURL, nil
+	}
+
+	if strings.HasPrefix(sourceURL, "https://") || strings.HasPrefix(sourceURL, "http://") || strings.HasPrefix(sourceURL, "ssh://") {
+		return "git", sourceURL, nil
+	}
+
+	return "", "", fmt.Errorf("unsupported source URL scheme (expected oci://, https://, http://, or ssh://): %s", sourceURL)
+}
+
+// parseRefSpec parses a reference specification string in the format "key1=value1,key2=value2".
+// Returns a map of key-value pairs.
+func parseRefSpec(refSpec string) (map[string]string, error) {
+	result := make(map[string]string)
+
+	refSpec = strings.TrimSpace(refSpec)
+	if refSpec == "" {
+		return result, nil
+	}
+
+	pairs := strings.Split(refSpec, ",")
+	for _, pair := range pairs {
+		pair = strings.TrimSpace(pair)
+		if pair == "" {
+			continue
+		}
+
+		// Split on first '=' only to allow '=' in values (e.g., digest=sha256:...)
+		idx := strings.Index(pair, "=")
+		if idx == -1 {
+			return nil, fmt.Errorf("invalid reference specification format: %q (expected key=value)", pair)
+		}
+
+		key := strings.TrimSpace(pair[:idx])
+		value := strings.TrimSpace(pair[idx+1:])
+
+		if key == "" {
+			return nil, fmt.Errorf("empty key in reference specification: %q", pair)
+		}
+		if value == "" {
+			return nil, fmt.Errorf("empty value for key %q in reference specification", key)
+		}
+
+		result[key] = value
+	}
+
+	return result, nil
+}
+
+// Valid reference keys for OCI repositories
+var validOCIRefKeys = map[string]bool{
+	"digest":       true,
+	"semver":       true,
+	"semverFilter": true,
+	"tag":          true,
+}
+
+// Valid reference keys for Git repositories
+var validGitRefKeys = map[string]bool{
+	"branch": true,
+	"tag":    true,
+	"semver": true,
+	"name":   true,
+	"commit": true,
+}
+
+// validateOCIRef validates reference keys for OCI repositories
+func validateOCIRef(refMap map[string]string) error {
+	for key := range refMap {
+		if !validOCIRefKeys[key] {
+			return fmt.Errorf("invalid OCI reference key %q (valid keys: digest, semver, semverFilter, tag)", key)
+		}
+	}
+
+	// Validate digest format if provided
+	if digest, ok := refMap["digest"]; ok {
+		if !strings.HasPrefix(digest, "sha256:") {
+			return fmt.Errorf("digest must be in format 'sha256:<hash>', got: %s", digest)
+		}
+	}
+
+	return nil
+}
+
+// validateGitRef validates reference keys for Git repositories
+func validateGitRef(refMap map[string]string) error {
+	for key := range refMap {
+		if !validGitRefKeys[key] {
+			return fmt.Errorf("invalid Git reference key %q (valid keys: branch, tag, semver, name, commit)", key)
+		}
+	}
+
+	// Validate commit format if provided (should be a hex string)
+	if commit, ok := refMap["commit"]; ok {
+		if len(commit) < 7 {
+			return fmt.Errorf("commit SHA should be at least 7 characters, got: %s", commit)
+		}
+		for _, c := range commit {
+			if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) {
+				return fmt.Errorf("commit SHA should be a hexadecimal string, got: %s", commit)
+			}
+		}
+	}
+
+	return nil
+}
+
+// generateOCIRepository creates an OCIRepository resource
+func generateOCIRepository(name, repoURL string, refMap map[string]string) (*sourcev1.OCIRepository, error) {
+	if err := validateOCIRef(refMap); err != nil {
+		return nil, err
+	}
+
+	obj := &sourcev1.OCIRepository{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: sourcev1.GroupVersion.String(),
+			Kind:       sourcev1.OCIRepositoryKind,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "cozy-system",
+		},
+		Spec: sourcev1.OCIRepositorySpec{
+			URL:      repoURL,
+			Interval: metav1.Duration{Duration: 5 * time.Minute},
+		},
+	}
+
+	// Set reference if any ref options are provided
+	if len(refMap) > 0 {
+		obj.Spec.Reference = &sourcev1.OCIRepositoryRef{
+			Digest:       refMap["digest"],
+			SemVer:       refMap["semver"],
+			SemverFilter: refMap["semverFilter"],
+			Tag:          refMap["tag"],
+		}
+	}
+
+	return obj, nil
+}
+
+// generateGitRepository creates a GitRepository resource
+func generateGitRepository(name, repoURL string, refMap map[string]string) (*sourcev1.GitRepository, error) {
+	if err := validateGitRef(refMap); err != nil {
+		return nil, err
+	}
+
+	obj := &sourcev1.GitRepository{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: sourcev1.GroupVersion.String(),
+			Kind:       sourcev1.GitRepositoryKind,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: "cozy-system",
+		},
+		Spec: sourcev1.GitRepositorySpec{
+			URL:      repoURL,
+			Interval: metav1.Duration{Duration: 5 * time.Minute},
+		},
+	}
+
+	// Set reference if any ref options are provided
+	if len(refMap) > 0 {
+		obj.Spec.Reference = &sourcev1.GitRepositoryRef{
+			Branch: refMap["branch"],
+			Tag:    refMap["tag"],
+			SemVer: refMap["semver"],
+			Name:   refMap["name"],
+			Commit: refMap["commit"],
+		}
+	}
+
+	return obj, nil
+}
--- a/go.mod
+++ b/go.mod
@ -6,6 +6,7 @@ go 1.25.0

 require (
 	github.com/fluxcd/helm-controller/api v1.4.3
+	github.com/fluxcd/source-controller/api v1.7.4
 	github.com/go-logr/logr v1.4.3
 	github.com/go-logr/zapr v1.3.0
 	github.com/google/gofuzz v1.2.0
@ -18,14 +19,14 @@ require (
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.34.1
 	k8s.io/apiextensions-apiserver v0.34.1
-	k8s.io/apimachinery v0.34.1
+	k8s.io/apimachinery v0.34.2
 	k8s.io/apiserver v0.34.1
 	k8s.io/client-go v0.34.1
 	k8s.io/component-base v0.34.1
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b
 	k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d
-	sigs.k8s.io/controller-runtime v0.22.2
+	sigs.k8s.io/controller-runtime v0.22.4
 	sigs.k8s.io/structured-merge-diff/v4 v4.7.0
 )

@ -44,8 +45,9 @@ require (
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/fluxcd/pkg/apis/acl v0.9.0 // indirect
 	github.com/fluxcd/pkg/apis/kustomize v1.13.0 // indirect
-	github.com/fluxcd/pkg/apis/meta v1.22.0 // indirect
+	github.com/fluxcd/pkg/apis/meta v1.23.0 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
--- a/go.sum
+++ b/go.sum
@ -37,10 +37,14 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fluxcd/helm-controller/api v1.4.3 h1:CdZwjL1liXmYCWyk2jscmFEB59tICIlnWB9PfDDW5q4=
 github.com/fluxcd/helm-controller/api v1.4.3/go.mod h1:0XrBhKEaqvxyDj/FziG1Q8Fmx2UATdaqLgYqmZh6wW4=
+github.com/fluxcd/pkg/apis/acl v0.9.0 h1:wBpgsKT+jcyZEcM//OmZr9RiF8klL3ebrDp2u2ThsnA=
+github.com/fluxcd/pkg/apis/acl v0.9.0/go.mod h1:TttNS+gocsGLwnvmgVi3/Yscwqrjc17+vhgYfqkfrV4=
 github.com/fluxcd/pkg/apis/kustomize v1.13.0 h1:GGf0UBVRIku+gebY944icVeEIhyg1P/KE3IrhOyJJnE=
 github.com/fluxcd/pkg/apis/kustomize v1.13.0/go.mod h1:TLKVqbtnzkhDuhWnAsN35977HvRfIjs+lgMuNro/LEc=
-github.com/fluxcd/pkg/apis/meta v1.22.0 h1:EHWQH5ZWml7i8eZ/AMjm1jxid3j/PQ31p+hIwCt6crM=
-github.com/fluxcd/pkg/apis/meta v1.22.0/go.mod h1:Kc1+bWe5p0doROzuV9XiTfV/oL3ddsemYXt8ZYWdVVg=
+github.com/fluxcd/pkg/apis/meta v1.23.0 h1:fLis5YcHnOsyKYptzBtituBm5EWNx13I0bXQsy0FG4s=
+github.com/fluxcd/pkg/apis/meta v1.23.0/go.mod h1:UWsIbBPCxYvoVklr2mV2uLFBf/n17dNAmKFjRfApdDo=
+github.com/fluxcd/source-controller/api v1.7.4 h1:+EOVnRA9LmLxOx7J273l7IOEU39m+Slt/nQGBy69ygs=
+github.com/fluxcd/source-controller/api v1.7.4/go.mod h1:ruf49LEgZRBfcP+eshl2n9SX1MfHayCcViAIGnZcaDY=
 github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
 github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM=
@ -312,8 +316,8 @@ k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d h1:wAhiDyZ4Tdtt7e46e9M5ZSAJ/MnPG
 k8s.io/utils v0.0.0-20250820121507-0af2bda4dd1d/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2 h1:jpcvIRr3GLoUoEKRkHKSmGjxb6lWwrBlJsXc+eUYQHM=
 sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.2/go.mod h1:Ve9uj1L+deCXFrPOk1LpFXqTg7LCFzFso6PA48q/XZw=
-sigs.k8s.io/controller-runtime v0.22.2 h1:cK2l8BGWsSWkXz09tcS4rJh95iOLney5eawcK5A33r4=
-sigs.k8s.io/controller-runtime v0.22.2/go.mod h1:+QX1XUpTXN4mLoblf4tqr5CQcyHPAki2HLXqQMY6vh8=
+sigs.k8s.io/controller-runtime v0.22.4 h1:GEjV7KV3TY8e+tJ2LCTxUTanW4z/FmNB7l327UfMq9A=
+sigs.k8s.io/controller-runtime v0.22.4/go.mod h1:+QX1XUpTXN4mLoblf4tqr5CQcyHPAki2HLXqQMY6vh8=
 sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
 sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/randfill v0.0.0-20250304075658-069ef1bbf016/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
--- a/internal/fluxinstall/install.go
+++ b/internal/fluxinstall/install.go
@ -0,0 +1,378 @@
+/*
+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"
+	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)
+
+	// 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 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, 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.containers
+		spec, found, err := unstructured.NestedMap(obj.Object, "spec", "template", "spec")
+		if !found {
+			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 []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
+}
+
--- a/internal/fluxinstall/manifests.embed.go
+++ b/internal/fluxinstall/manifests.embed.go
@ -0,0 +1,51 @@
+/*
+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 (
+	"embed"
+	"fmt"
+	"io/fs"
+	"os"
+	"path"
+)
+
+//go:embed manifests/*.yaml
+var embeddedFluxManifests embed.FS
+
+// WriteEmbeddedManifests extracts embedded Flux manifests to a temporary directory.
+func WriteEmbeddedManifests(dir string) error {
+	manifests, err := fs.ReadDir(embeddedFluxManifests, "manifests")
+	if err != nil {
+		return fmt.Errorf("failed to read embedded manifests: %w", err)
+	}
+
+	for _, manifest := range manifests {
+		data, err := fs.ReadFile(embeddedFluxManifests, path.Join("manifests", manifest.Name()))
+		if err != nil {
+			return fmt.Errorf("failed to read file %s: %w", manifest.Name(), err)
+		}
+
+		outputPath := path.Join(dir, manifest.Name())
+		if err := os.WriteFile(outputPath, data, 0666); err != nil {
+			return fmt.Errorf("failed to write file %s: %w", outputPath, err)
+		}
+	}
+
+	return nil
+}
+
--- a/internal/fluxinstall/manifests/fluxcd.yaml
+++ b/internal/fluxinstall/manifests/fluxcd.yaml
--- a/packages/core/flux-aio/Makefile
+++ b/packages/core/flux-aio/Makefile
@ -12,7 +12,10 @@ apply:
 diff:
 	cozypkg show -n $(NAMESPACE) $(NAME) --plain | kubectl diff -f-

-update:
+update: update-old update-new
+
+# TODO: remove old manifest after migration to cozystack-operator
+update-old:
 	timoni bundle build -f flux-aio.cue > templates/fluxcd.yaml
 	yq eval '(select(.kind == "Namespace") | .metadata.labels."pod-security.kubernetes.io/enforce") = "privileged"' -i templates/fluxcd.yaml
 	sed -i templates/fluxcd.yaml \
@ -20,3 +23,12 @@ update:
 		-e 's|\.cluster\.local\.,||g' -e 's|\.cluster\.local\,||g' -e 's|\.cluster\.local\.||g' \
 		-e '/value: .svc/a \            {{- include "cozy.kubernetes_envs" . | nindent 12 }}' \
 		-e '/hostNetwork: true/i \      dnsPolicy: ClusterFirstWithHostNet'
+
+update-new:
+	timoni bundle build -f flux-aio.cue > ../../../internal/fluxinstall/manifests/fluxcd.yaml
+	yq eval '(select(.kind == "Namespace") | .metadata.labels."pod-security.kubernetes.io/enforce") = "privileged"' -i ../../../internal/fluxinstall/manifests/fluxcd.yaml
+	sed -i ../../../internal/fluxinstall/manifests/fluxcd.yaml \
+		-e '/timoni/d' \
+		-e 's|\.cluster\.local\.,||g' -e 's|\.cluster\.local\,||g' -e 's|\.cluster\.local\.||g'
+	# TODO: solve dns issue with hostNetwork for installing helmreleases in tenant k8s clusters
+	#-e '/hostNetwork: true/i \      dnsPolicy: ClusterFirstWithHostNet'