mirror of
https://github.com/cozystack/cozystack.git
synced 2026-11-03 18:47:04 +09:00
Add first-class fields so a package/app can declare the Flux helm-controller v1.5.0 kstatus health gate for the custom resource it wraps, so HelmRelease.Ready means "the CR is actually healthy" instead of "helm applied the CR" (#2642). API-owner approved in #3263. - api: WaitStrategy (scalar, maps to HelmReleaseSpec.WaitStrategy.Name — a deliberate simplification) + HealthCheckExprs ([]kustomize.CustomHealthCheck, the upstream type verbatim) on both ApplicationDefinition.spec.release (the apiserver path) and ComponentInstall (the operator path). - config.ResolveWaitStrategy: shared by both HelmRelease builders. Couples the default — healthCheckExprs imply the poller strategy when none is set, since exprs are only evaluated under poller. This removes the "must also set waitStrategy: poller" footgun and does not depend on the controller's global default. - thread through buildHelmReleaseSpec (operator) and convertApplicationToHelmRelease (apiserver); the apiserver loads the typed spec.release fields in start.go. - regenerate deepcopy, the ApplicationDefinition and PackageSource CRDs, and openapi. The backupstrategy CRDs also regenerate — benign k8s-0.35 PodSpec description drift, not a functional change. disableWait is intentionally not added (YAGNI): the only known need (the kubernetes tenant-addon deadlock) was resolved child-side, and waitStrategy: legacy is not equivalent (legacy still waits Helm-v3-style; disableWait skips waiting entirely). Add it later when a concrete leaf needs a true no-gate escape hatch. Refs #2642 Assisted-By: Claude <noreply@anthropic.com> Signed-off-by: Myasnikov Daniil <myasnikovdaniil2001@gmail.com>
1912 lines
70 KiB
Go
1912 lines
70 KiB
Go
/*
|
|
Copyright 2024 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 application
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
cozyv1alpha1 "github.com/cozystack/cozystack/api/v1alpha1"
|
|
helmv2 "github.com/fluxcd/helm-controller/api/v2"
|
|
corev1 "k8s.io/api/core/v1"
|
|
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/fields"
|
|
labels "k8s.io/apimachinery/pkg/labels"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
|
"k8s.io/apimachinery/pkg/selection"
|
|
"k8s.io/apimachinery/pkg/util/duration"
|
|
"k8s.io/apimachinery/pkg/util/validation/field"
|
|
"k8s.io/apimachinery/pkg/watch"
|
|
"k8s.io/apiserver/pkg/endpoints/request"
|
|
"k8s.io/apiserver/pkg/registry/rest"
|
|
"k8s.io/client-go/util/retry"
|
|
"k8s.io/klog/v2"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
|
|
fluxshard "github.com/cozystack/cozystack/internal/fluxshardoperator"
|
|
"github.com/cozystack/cozystack/pkg/apis/apps/presets"
|
|
appsv1alpha1 "github.com/cozystack/cozystack/pkg/apis/apps/v1alpha1"
|
|
"github.com/cozystack/cozystack/pkg/apis/apps/validation"
|
|
"github.com/cozystack/cozystack/pkg/config"
|
|
"github.com/cozystack/cozystack/pkg/registry"
|
|
fieldfilter "github.com/cozystack/cozystack/pkg/registry/fields"
|
|
"github.com/cozystack/cozystack/pkg/registry/sorting"
|
|
internalapiext "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions"
|
|
apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
|
|
structuralschema "k8s.io/apiextensions-apiserver/pkg/apiserver/schema"
|
|
|
|
// Importing API errors package to construct appropriate error responses
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
)
|
|
|
|
// Ensure REST implements necessary interfaces
|
|
var (
|
|
_ rest.Getter = &REST{}
|
|
_ rest.Lister = &REST{}
|
|
_ rest.Updater = &REST{}
|
|
_ rest.Creater = &REST{}
|
|
_ rest.GracefulDeleter = &REST{}
|
|
_ rest.Watcher = &REST{}
|
|
_ rest.Patcher = &REST{}
|
|
)
|
|
|
|
// Define constants for label and annotation prefixes
|
|
const (
|
|
LabelPrefix = "apps.cozystack.io-"
|
|
AnnotationPrefix = "apps.cozystack.io-"
|
|
)
|
|
|
|
// Application label keys - use constants from API package
|
|
const (
|
|
ApplicationKindLabel = appsv1alpha1.ApplicationKindLabel
|
|
ApplicationGroupLabel = appsv1alpha1.ApplicationGroupLabel
|
|
ApplicationNameLabel = appsv1alpha1.ApplicationNameLabel
|
|
)
|
|
|
|
// REST implements the RESTStorage interface for Application resources
|
|
type REST struct {
|
|
c client.Client
|
|
w client.WithWatch
|
|
gvr schema.GroupVersionResource
|
|
gvk schema.GroupVersionKind
|
|
kindName string
|
|
singularName string
|
|
releaseConfig config.ReleaseConfig
|
|
specSchema *structuralschema.Structural
|
|
}
|
|
|
|
// buildSpecSchema parses an OpenAPI-v3 JSON schema string and returns the
|
|
// structural-schema form used for defaulting. Returns (nil, nil) when raw
|
|
// is empty after trimming. Returns (nil, error) for malformed JSON,
|
|
// v1→internal conversion failures, or structural-schema construction
|
|
// errors — callers can decide whether to log-and-continue (NewREST does)
|
|
// or fail hard (tests do).
|
|
func buildSpecSchema(raw string) (*structuralschema.Structural, error) {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" {
|
|
return nil, nil
|
|
}
|
|
var v1js apiextv1.JSONSchemaProps
|
|
if err := json.Unmarshal([]byte(raw), &v1js); err != nil {
|
|
return nil, fmt.Errorf("unmarshal v1 OpenAPI schema: %w", err)
|
|
}
|
|
scheme := runtime.NewScheme()
|
|
_ = internalapiext.AddToScheme(scheme)
|
|
_ = apiextv1.AddToScheme(scheme)
|
|
var ijs internalapiext.JSONSchemaProps
|
|
if err := scheme.Convert(&v1js, &ijs, nil); err != nil {
|
|
return nil, fmt.Errorf("convert v1->internal JSONSchemaProps: %w", err)
|
|
}
|
|
s, err := structuralschema.NewStructural(&ijs)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build structural schema: %w", err)
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// NewREST creates a new REST storage for Application with specific configuration
|
|
func NewREST(c client.Client, w client.WithWatch, config *config.Resource) *REST {
|
|
specSchema, err := buildSpecSchema(config.Application.OpenAPISchema)
|
|
if err != nil {
|
|
klog.Errorf("Failed to build spec schema: %v", err)
|
|
}
|
|
|
|
return &REST{
|
|
c: c,
|
|
w: w,
|
|
gvr: schema.GroupVersionResource{
|
|
Group: appsv1alpha1.GroupName,
|
|
Version: "v1alpha1",
|
|
Resource: config.Application.Plural,
|
|
},
|
|
gvk: schema.GroupVersion{
|
|
Group: appsv1alpha1.GroupName,
|
|
Version: "v1alpha1",
|
|
}.WithKind(config.Application.Kind),
|
|
kindName: config.Application.Kind,
|
|
singularName: config.Application.Singular,
|
|
releaseConfig: config.Release,
|
|
specSchema: specSchema,
|
|
}
|
|
}
|
|
|
|
// NamespaceScoped indicates whether the resource is namespaced
|
|
func (r *REST) NamespaceScoped() bool {
|
|
return true
|
|
}
|
|
|
|
// GetSingularName returns the singular name of the resource
|
|
func (r *REST) GetSingularName() string {
|
|
return r.singularName
|
|
}
|
|
|
|
// Create handles the creation of a new Application by converting it to a HelmRelease
|
|
func (r *REST) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
|
|
// Assert the object is of type Application
|
|
app, ok := obj.(*appsv1alpha1.Application)
|
|
if !ok {
|
|
return nil, fmt.Errorf("expected *appsv1alpha1.Application object, got %T", obj)
|
|
}
|
|
|
|
// Validate Application name format (DNS-1035 plus any kind-specific rules)
|
|
if errs := r.validateNameFormat(app.Name); len(errs) > 0 {
|
|
return nil, apierrors.NewInvalid(r.gvk.GroupKind(), app.Name, errs)
|
|
}
|
|
|
|
// Validate name length against Helm release and label limits
|
|
if nameLenErrs := r.validateNameLength(app.Name); len(nameLenErrs) > 0 {
|
|
return nil, apierrors.NewInvalid(r.gvk.GroupKind(), app.Name, nameLenErrs)
|
|
}
|
|
|
|
// For Tenant applications, also validate that the computed workload
|
|
// namespace fits within the DNS-1123 label limit. A deeply-nested tenant
|
|
// can exceed the limit even when its own name passes the per-name Helm
|
|
// release length check, because the namespace is formed from the full
|
|
// ancestor chain.
|
|
if r.kindName == "Tenant" {
|
|
if nsErrs := r.validateTenantNamespaceLength(app.Namespace, app.Name); len(nsErrs) > 0 {
|
|
return nil, apierrors.NewInvalid(r.gvk.GroupKind(), app.Name, nsErrs)
|
|
}
|
|
// Enforce hierarchical quota allocation: a child tenant's declared
|
|
// quota may not exceed its parent's remaining (un-carved) quota.
|
|
if qErrs := r.validateTenantResourceQuotas(ctx, app); len(qErrs) > 0 {
|
|
return nil, apierrors.NewInvalid(r.gvk.GroupKind(), app.Name, qErrs)
|
|
}
|
|
}
|
|
|
|
// Validate that values don't contain reserved keys (starting with "_")
|
|
if err := validateNoInternalKeys(app.Spec); err != nil {
|
|
return nil, apierrors.NewBadRequest(err.Error())
|
|
}
|
|
|
|
r.warnLegacyPresets(app)
|
|
|
|
// Run the genericapiserver-supplied validating admission chain
|
|
// (validating webhooks + ValidatingAdmissionPolicies) before
|
|
// persisting anything. Mutating webhooks already ran upstream in
|
|
// genericapiserver/handlers.create before this REST handler was
|
|
// invoked; createValidation is documented as validate-only (must
|
|
// not transform the object). Custom REST handlers must invoke
|
|
// this hook explicitly — unlike genericregistry.Store, which
|
|
// wires it automatically.
|
|
if createValidation != nil {
|
|
if err := createValidation(ctx, obj); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// Convert Application to HelmRelease
|
|
helmRelease, err := r.ConvertApplicationToHelmRelease(app)
|
|
if err != nil {
|
|
klog.Errorf("Conversion error: %v", err)
|
|
return nil, fmt.Errorf("conversion error: %v", err)
|
|
}
|
|
|
|
// Merge system labels (from config) directly
|
|
helmRelease.Labels = mergeMaps(r.releaseConfig.Labels, helmRelease.Labels)
|
|
// Merge user labels with prefix
|
|
helmRelease.Labels = mergeMaps(helmRelease.Labels, addPrefixedMap(app.Labels, LabelPrefix))
|
|
// Add application metadata labels
|
|
if helmRelease.Labels == nil {
|
|
helmRelease.Labels = make(map[string]string)
|
|
}
|
|
helmRelease.Labels[ApplicationKindLabel] = r.kindName
|
|
helmRelease.Labels[ApplicationGroupLabel] = r.gvk.Group
|
|
helmRelease.Labels[ApplicationNameLabel] = app.Name
|
|
// Note: Annotations from config are not handled as r.releaseConfig.Annotations is undefined
|
|
|
|
klog.V(6).Infof("Creating HelmRelease %s in namespace %s", helmRelease.Name, app.Namespace)
|
|
|
|
// Create HelmRelease in Kubernetes
|
|
err = r.c.Create(ctx, helmRelease, &client.CreateOptions{Raw: options})
|
|
if err != nil {
|
|
klog.Errorf("Failed to create HelmRelease %s: %v", helmRelease.Name, err)
|
|
return nil, fmt.Errorf("failed to create HelmRelease: %v", err)
|
|
}
|
|
|
|
// Convert the created HelmRelease back to Application
|
|
convertedApp, err := r.ConvertHelmReleaseToApplication(ctx, helmRelease)
|
|
if err != nil {
|
|
klog.Errorf("Conversion error from HelmRelease to Application for resource %s: %v", helmRelease.GetName(), err)
|
|
return nil, fmt.Errorf("conversion error: %v", err)
|
|
}
|
|
|
|
klog.V(6).Infof("Successfully created and converted HelmRelease %s to Application", helmRelease.GetName())
|
|
|
|
klog.V(6).Infof("Successfully retrieved and converted resource %s of type %s", convertedApp.GetName(), r.gvr.Resource)
|
|
return &convertedApp, nil
|
|
}
|
|
|
|
// Get retrieves an Application by converting the corresponding HelmRelease
|
|
func (r *REST) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
|
|
namespace, err := r.getNamespace(ctx)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get namespace: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(6).Infof("Attempting to retrieve resource %s of type %s in namespace %s", name, r.gvr.Resource, namespace)
|
|
|
|
// Get the corresponding HelmRelease using the new prefix
|
|
helmReleaseName := r.releaseConfig.Prefix + name
|
|
helmRelease := &helmv2.HelmRelease{}
|
|
err = r.c.Get(ctx, client.ObjectKey{Namespace: namespace, Name: helmReleaseName}, helmRelease, &client.GetOptions{Raw: options})
|
|
if err != nil {
|
|
klog.Errorf("Error retrieving HelmRelease for resource %s: %v", name, err)
|
|
|
|
// Check if the error is a NotFound error
|
|
if apierrors.IsNotFound(err) {
|
|
// Return a NotFound error for the Application resource instead of HelmRelease
|
|
return nil, apierrors.NewNotFound(r.gvr.GroupResource(), name)
|
|
}
|
|
|
|
// For other errors, return them as-is
|
|
return nil, err
|
|
}
|
|
|
|
// Check if HelmRelease has required labels
|
|
if !r.hasRequiredApplicationLabels(helmRelease) {
|
|
klog.Errorf("HelmRelease %s does not match the required application labels", helmReleaseName)
|
|
// Return a NotFound error for the Application resource
|
|
return nil, apierrors.NewNotFound(r.gvr.GroupResource(), name)
|
|
}
|
|
|
|
// Convert HelmRelease to Application
|
|
convertedApp, err := r.ConvertHelmReleaseToApplication(ctx, helmRelease)
|
|
if err != nil {
|
|
klog.Errorf("Conversion error from HelmRelease to Application for resource %s: %v", name, err)
|
|
return nil, fmt.Errorf("conversion error: %v", err)
|
|
}
|
|
|
|
klog.V(6).Infof("Successfully retrieved and converted resource %s of kind %s", name, r.gvr.Resource)
|
|
return &convertedApp, nil
|
|
}
|
|
|
|
// List retrieves a list of Applications by converting HelmReleases
|
|
func (r *REST) List(ctx context.Context, options *metainternalversion.ListOptions) (runtime.Object, error) {
|
|
namespace, err := r.getNamespace(ctx)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get namespace: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(6).Infof("List called for %s in namespace %q", r.kindName, namespace)
|
|
|
|
// Get resource name from the request (if any)
|
|
var resourceName string
|
|
if requestInfo, ok := request.RequestInfoFrom(ctx); ok {
|
|
resourceName = requestInfo.Name
|
|
}
|
|
|
|
// Initialize variables for selector mapping
|
|
var helmLabelSelector labels.Selector
|
|
|
|
// Parse field selector for manual filtering
|
|
// controller-runtime cache doesn't support field selectors
|
|
// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
|
|
fieldFilter, err := fieldfilter.ParseFieldSelector(options.FieldSelector)
|
|
if err != nil {
|
|
klog.Errorf("Error parsing field selector: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
// If field selector specifies namespace different from context, return empty list
|
|
if fieldFilter.Namespace != "" && namespace != "" && namespace != fieldFilter.Namespace {
|
|
klog.V(6).Infof("Field selector namespace %s doesn't match context namespace %s, returning empty list", fieldFilter.Namespace, namespace)
|
|
return &appsv1alpha1.ApplicationList{
|
|
TypeMeta: metav1.TypeMeta{
|
|
APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
|
|
Kind: r.kindName + "List",
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// Convert Application name to HelmRelease name for manual filtering
|
|
var filterByName string
|
|
if fieldFilter.Name != "" {
|
|
filterByName = r.releaseConfig.Prefix + fieldFilter.Name
|
|
}
|
|
|
|
// Process label.selector
|
|
// Always add application metadata label requirements
|
|
appKindReq, err := labels.NewRequirement(ApplicationKindLabel, selection.Equals, []string{r.kindName})
|
|
if err != nil {
|
|
klog.Errorf("Error creating application kind label requirement: %v", err)
|
|
return nil, fmt.Errorf("error creating application kind label requirement: %v", err)
|
|
}
|
|
appGroupReq, err := labels.NewRequirement(ApplicationGroupLabel, selection.Equals, []string{r.gvk.Group})
|
|
if err != nil {
|
|
klog.Errorf("Error creating application group label requirement: %v", err)
|
|
return nil, fmt.Errorf("error creating application group label requirement: %v", err)
|
|
}
|
|
labelRequirements := []labels.Requirement{*appKindReq, *appGroupReq}
|
|
|
|
if options.LabelSelector != nil {
|
|
ls := options.LabelSelector.String()
|
|
parsedLabels, err := labels.Parse(ls)
|
|
if err != nil {
|
|
klog.Errorf("Invalid label selector: %v", err)
|
|
return nil, fmt.Errorf("invalid label selector: %v", err)
|
|
}
|
|
if !parsedLabels.Empty() {
|
|
reqs, _ := parsedLabels.Requirements()
|
|
var prefixedReqs []labels.Requirement
|
|
for _, req := range reqs {
|
|
// Add prefix to each label key
|
|
prefixedReq, err := labels.NewRequirement(LabelPrefix+req.Key(), req.Operator(), req.Values().List())
|
|
if err != nil {
|
|
klog.Errorf("Error prefixing label key: %v", err)
|
|
return nil, fmt.Errorf("error prefixing label key: %v", err)
|
|
}
|
|
prefixedReqs = append(prefixedReqs, *prefixedReq)
|
|
}
|
|
labelRequirements = append(labelRequirements, prefixedReqs...)
|
|
}
|
|
}
|
|
helmLabelSelector = labels.NewSelector().Add(labelRequirements...)
|
|
|
|
klog.V(6).Infof("Using label selector: %s for kind: %s, group: %s", helmLabelSelector, r.kindName, r.gvk.Group)
|
|
|
|
// List HelmReleases with label selector only
|
|
// Field selectors are not supported by controller-runtime cache, so we filter manually below
|
|
hrList := &helmv2.HelmReleaseList{}
|
|
err = r.c.List(ctx, hrList, &client.ListOptions{
|
|
Namespace: namespace,
|
|
LabelSelector: helmLabelSelector,
|
|
})
|
|
if err != nil {
|
|
klog.Errorf("Error listing HelmReleases: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(6).Infof("Found %d HelmReleases with label selector", len(hrList.Items))
|
|
|
|
// Initialize Application items array
|
|
items := make([]appsv1alpha1.Application, 0, len(hrList.Items))
|
|
|
|
// Iterate over HelmReleases and convert to Applications
|
|
// Note: All HelmReleases already match the required labels due to server-side label selector filtering
|
|
for i := range hrList.Items {
|
|
hr := &hrList.Items[i]
|
|
|
|
// Apply manual field selector filtering (metadata.name and metadata.namespace)
|
|
// controller-runtime cache doesn't support field selectors
|
|
// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
|
|
if filterByName != "" && hr.Name != filterByName {
|
|
continue
|
|
}
|
|
if !fieldFilter.MatchesNamespace(hr.Namespace) {
|
|
continue
|
|
}
|
|
|
|
app, err := r.ConvertHelmReleaseToApplication(ctx, hr)
|
|
if err != nil {
|
|
klog.Errorf("Error converting HelmRelease %s to Application: %v", hr.GetName(), err)
|
|
continue
|
|
}
|
|
|
|
// If resourceName is set, check for match
|
|
if resourceName != "" && app.Name != resourceName {
|
|
continue
|
|
}
|
|
|
|
// Apply label.selector
|
|
if options.LabelSelector != nil {
|
|
sel, err := labels.Parse(options.LabelSelector.String())
|
|
if err != nil {
|
|
klog.Errorf("Invalid label selector: %v", err)
|
|
continue
|
|
}
|
|
if !sel.Matches(labels.Set(app.Labels)) {
|
|
continue
|
|
}
|
|
}
|
|
|
|
// Apply field.selector by name and namespace (if specified)
|
|
if options.FieldSelector != nil {
|
|
fs, err := fields.ParseSelector(options.FieldSelector.String())
|
|
if err != nil {
|
|
klog.Errorf("Invalid field selector: %v", err)
|
|
continue
|
|
}
|
|
fieldsSet := fields.Set{
|
|
"metadata.name": app.Name,
|
|
"metadata.namespace": app.Namespace,
|
|
}
|
|
if !fs.Matches(fieldsSet) {
|
|
continue
|
|
}
|
|
}
|
|
|
|
items = append(items, app)
|
|
}
|
|
|
|
// Create ApplicationList with proper kind
|
|
appList := r.NewList().(*appsv1alpha1.ApplicationList)
|
|
|
|
// Get ResourceVersion from list or compute from items
|
|
// controller-runtime cached client may not set ResourceVersion on the list itself
|
|
listRV := hrList.GetResourceVersion()
|
|
if listRV == "" {
|
|
listRV, _ = registry.MaxResourceVersion(hrList)
|
|
}
|
|
appList.SetResourceVersion(listRV)
|
|
appList.Items = items
|
|
|
|
sorting.ByNamespacedName[appsv1alpha1.Application, *appsv1alpha1.Application](appList.Items)
|
|
|
|
klog.V(6).Infof("List returning %d items for %s in namespace %q, resourceVersion=%q",
|
|
len(items), r.kindName, namespace, appList.GetResourceVersion())
|
|
return appList, nil
|
|
}
|
|
|
|
// Update updates an existing Application by converting it to a HelmRelease
|
|
func (r *REST) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
|
|
// Retrieve the existing Application
|
|
oldObj, err := r.Get(ctx, name, &metav1.GetOptions{})
|
|
if err != nil {
|
|
if apierrors.IsNotFound(err) {
|
|
if !forceAllowCreate {
|
|
return nil, false, err
|
|
}
|
|
// If not found and force allow create, create a new one
|
|
obj, err := objInfo.UpdatedObject(ctx, nil)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get updated object: %v", err)
|
|
return nil, false, err
|
|
}
|
|
createdObj, err := r.Create(ctx, obj, createValidation, &metav1.CreateOptions{})
|
|
if err != nil {
|
|
klog.Errorf("Failed to create new Application: %v", err)
|
|
return nil, false, err
|
|
}
|
|
return createdObj, true, nil
|
|
}
|
|
klog.Errorf("Failed to get existing Application %s: %v", name, err)
|
|
return nil, false, err
|
|
}
|
|
|
|
// Update the Application object
|
|
newObj, err := objInfo.UpdatedObject(ctx, oldObj)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get updated object: %v", err)
|
|
return nil, false, err
|
|
}
|
|
|
|
// Validate the update if a validation function is provided
|
|
if updateValidation != nil {
|
|
if err := updateValidation(ctx, newObj, oldObj); err != nil {
|
|
klog.Errorf("Update validation failed for Application %s: %v", name, err)
|
|
return nil, false, err
|
|
}
|
|
}
|
|
|
|
// Assert the new object is of type Application
|
|
app, ok := newObj.(*appsv1alpha1.Application)
|
|
if !ok {
|
|
klog.Errorf("expected *appsv1alpha1.Application object, got %T", newObj)
|
|
return nil, false, fmt.Errorf("expected *appsv1alpha1.Application object, got %T", newObj)
|
|
}
|
|
|
|
// Note: name validation (DNS-1035 format + length) is intentionally skipped on
|
|
// Update because Kubernetes names are immutable. Validating here would block
|
|
// updates to pre-existing resources whose names don't conform to the new rules.
|
|
|
|
// Validate that values don't contain reserved keys (starting with "_")
|
|
if err := validateNoInternalKeys(app.Spec); err != nil {
|
|
return nil, false, apierrors.NewBadRequest(err.Error())
|
|
}
|
|
|
|
// Enforce hierarchical quota allocation on quota changes too: raising a
|
|
// child tenant's declared quota above its parent's remaining budget is
|
|
// rejected the same way as on create.
|
|
if r.kindName == "Tenant" {
|
|
if qErrs := r.validateTenantResourceQuotas(ctx, app); len(qErrs) > 0 {
|
|
return nil, false, apierrors.NewInvalid(r.gvk.GroupKind(), app.Name, qErrs)
|
|
}
|
|
}
|
|
|
|
r.warnLegacyPresets(app)
|
|
|
|
// Convert Application to HelmRelease
|
|
helmRelease, err := r.ConvertApplicationToHelmRelease(app)
|
|
if err != nil {
|
|
klog.Errorf("Conversion error: %v", err)
|
|
return nil, false, fmt.Errorf("conversion error: %v", err)
|
|
}
|
|
|
|
// Fetch the live HelmRelease: it backs the ResourceVersion when the
|
|
// converted object carries none, and runtime-managed labels are carried
|
|
// over from it below.
|
|
cur := &helmv2.HelmRelease{}
|
|
if err := r.c.Get(ctx, client.ObjectKey{Namespace: helmRelease.Namespace, Name: helmRelease.Name}, cur, &client.GetOptions{Raw: &metav1.GetOptions{}}); err != nil {
|
|
return nil, false, fmt.Errorf("failed to fetch current HelmRelease: %w", err)
|
|
}
|
|
if helmRelease.ResourceVersion == "" {
|
|
helmRelease.SetResourceVersion(cur.GetResourceVersion())
|
|
}
|
|
|
|
// Merge system labels (from config) directly
|
|
helmRelease.Labels = mergeMaps(r.releaseConfig.Labels, helmRelease.Labels)
|
|
// Merge user labels with prefix
|
|
helmRelease.Labels = mergeMaps(helmRelease.Labels, addPrefixedMap(app.Labels, LabelPrefix))
|
|
// Add application metadata labels
|
|
if helmRelease.Labels == nil {
|
|
helmRelease.Labels = make(map[string]string)
|
|
}
|
|
helmRelease.Labels[ApplicationKindLabel] = r.kindName
|
|
helmRelease.Labels[ApplicationGroupLabel] = r.gvk.Group
|
|
helmRelease.Labels[ApplicationNameLabel] = app.Name
|
|
// Note: Annotations from config are not handled as r.releaseConfig.Annotations is undefined
|
|
|
|
// The flux-shard-operator assigns each tenant HelmRelease to a
|
|
// helm-controller shard by rewriting this label at runtime. Rebuilding
|
|
// the object from the Application reverts it to the ApplicationDefinition
|
|
// default, which would bounce the HelmRelease off its shard on every
|
|
// update, so the live value wins.
|
|
if shard, ok := cur.Labels[fluxshard.ShardKeyLabel]; ok {
|
|
helmRelease.Labels[fluxshard.ShardKeyLabel] = shard
|
|
}
|
|
|
|
klog.V(6).Infof("Updating HelmRelease %s in namespace %s", helmRelease.Name, helmRelease.Namespace)
|
|
|
|
// Update the HelmRelease in Kubernetes.
|
|
//
|
|
// Flux's helm-controller (and other controllers) continuously write the
|
|
// HelmRelease's status, which shares the object's resourceVersion. When a
|
|
// caller updates an app CR while a prior reconcile is still in flight, the
|
|
// resourceVersion read above goes stale and the Update is rejected with a
|
|
// 409 Conflict. The HelmRelease spec is fully derived from the Application
|
|
// the caller just applied, so a stale-resourceVersion conflict is never a
|
|
// real spec conflict here: refresh the resourceVersion from the live object
|
|
// and retry.
|
|
err = retry.RetryOnConflict(retry.DefaultBackoff, func() error {
|
|
updateErr := r.c.Update(ctx, helmRelease, &client.UpdateOptions{Raw: &metav1.UpdateOptions{}})
|
|
if apierrors.IsConflict(updateErr) {
|
|
cur := &helmv2.HelmRelease{}
|
|
if getErr := r.c.Get(ctx, client.ObjectKey{Namespace: helmRelease.Namespace, Name: helmRelease.Name}, cur, &client.GetOptions{Raw: &metav1.GetOptions{}}); getErr != nil {
|
|
return getErr
|
|
}
|
|
helmRelease.SetResourceVersion(cur.GetResourceVersion())
|
|
}
|
|
return updateErr
|
|
})
|
|
if err != nil {
|
|
klog.Errorf("Failed to update HelmRelease %s: %v", helmRelease.Name, err)
|
|
return nil, false, fmt.Errorf("failed to update HelmRelease: %v", err)
|
|
}
|
|
|
|
// Convert the updated HelmRelease back to Application
|
|
convertedApp, err := r.ConvertHelmReleaseToApplication(ctx, helmRelease)
|
|
if err != nil {
|
|
klog.Errorf("Conversion error from HelmRelease to Application for resource %s: %v", helmRelease.GetName(), err)
|
|
return nil, false, fmt.Errorf("conversion error: %v", err)
|
|
}
|
|
|
|
klog.V(6).Infof("Successfully updated and converted HelmRelease %s to Application", helmRelease.GetName())
|
|
|
|
klog.V(6).Infof("Returning updated Application object: %+v", convertedApp)
|
|
|
|
return &convertedApp, false, nil
|
|
}
|
|
|
|
// Delete removes an Application by deleting the corresponding HelmRelease
|
|
func (r *REST) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
|
|
namespace, err := r.getNamespace(ctx)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get namespace: %v", err)
|
|
return nil, false, err
|
|
}
|
|
|
|
klog.V(6).Infof("Attempting to delete HelmRelease %s in namespace %s", name, namespace)
|
|
|
|
// Construct HelmRelease name with the configured prefix
|
|
helmReleaseName := r.releaseConfig.Prefix + name
|
|
|
|
// Retrieve the HelmRelease before attempting to delete
|
|
helmRelease := &helmv2.HelmRelease{}
|
|
err = r.c.Get(ctx, client.ObjectKey{Namespace: namespace, Name: helmReleaseName}, helmRelease, &client.GetOptions{Raw: &metav1.GetOptions{}})
|
|
if err != nil {
|
|
if apierrors.IsNotFound(err) {
|
|
// If HelmRelease does not exist, return NotFound error for Application
|
|
klog.Errorf("HelmRelease %s not found in namespace %s", helmReleaseName, namespace)
|
|
return nil, false, apierrors.NewNotFound(r.gvr.GroupResource(), name)
|
|
}
|
|
// For other errors, log and return
|
|
klog.Errorf("Error retrieving HelmRelease %s: %v", helmReleaseName, err)
|
|
return nil, false, err
|
|
}
|
|
|
|
// Validate that the HelmRelease has required labels
|
|
if !r.hasRequiredApplicationLabelsWithName(helmRelease, name) {
|
|
klog.Errorf("HelmRelease %s does not match the required application labels", helmReleaseName)
|
|
// Return NotFound error for Application resource
|
|
return nil, false, apierrors.NewNotFound(r.gvr.GroupResource(), name)
|
|
}
|
|
|
|
// Run the genericapiserver-supplied admission chain (validating webhooks
|
|
// and ValidatingAdmissionPolicies) on the resolved Application before
|
|
// removing the underlying HelmRelease. Custom REST handlers must invoke
|
|
// this hook explicitly — unlike genericregistry.Store, which wires it
|
|
// automatically.
|
|
if deleteValidation != nil {
|
|
converted, convErr := r.ConvertHelmReleaseToApplication(ctx, helmRelease)
|
|
if convErr != nil {
|
|
klog.Errorf("Failed to convert HelmRelease %s to Application for delete admission: %v", helmReleaseName, convErr)
|
|
return nil, false, convErr
|
|
}
|
|
if err := deleteValidation(ctx, &converted); err != nil {
|
|
return nil, false, err
|
|
}
|
|
}
|
|
|
|
klog.V(6).Infof("Deleting HelmRelease %s in namespace %s", helmReleaseName, namespace)
|
|
|
|
// Delete the HelmRelease corresponding to the Application
|
|
err = r.c.Delete(ctx, helmRelease, &client.DeleteOptions{Raw: options})
|
|
if err != nil {
|
|
klog.Errorf("Failed to delete HelmRelease %s: %v", helmReleaseName, err)
|
|
return nil, false, fmt.Errorf("failed to delete HelmRelease: %v", err)
|
|
}
|
|
|
|
klog.V(6).Infof("Successfully deleted HelmRelease %s", helmReleaseName)
|
|
return nil, true, nil
|
|
}
|
|
|
|
// Watch sets up a watch on HelmReleases, filters them based on application labels, and converts events to Applications
|
|
func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
|
|
namespace, err := r.getNamespace(ctx)
|
|
if err != nil {
|
|
klog.Errorf("Failed to get namespace: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
klog.V(6).Infof("Watch called for %s in namespace %q, resourceVersion=%q",
|
|
r.kindName, namespace, options.ResourceVersion)
|
|
|
|
// Get request information, including resource name if specified
|
|
var resourceName string
|
|
if requestInfo, ok := request.RequestInfoFrom(ctx); ok {
|
|
resourceName = requestInfo.Name
|
|
}
|
|
|
|
// Initialize variables for selector mapping
|
|
var helmLabelSelector labels.Selector
|
|
|
|
// Parse field selector for manual filtering
|
|
// controller-runtime cache doesn't support field selectors
|
|
// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
|
|
fieldFilter, err := fieldfilter.ParseFieldSelector(options.FieldSelector)
|
|
if err != nil {
|
|
klog.Errorf("Error parsing field selector: %v", err)
|
|
return nil, err
|
|
}
|
|
|
|
// Convert Application name to HelmRelease name for manual filtering
|
|
var filterByName string
|
|
if fieldFilter.Name != "" {
|
|
filterByName = r.releaseConfig.Prefix + fieldFilter.Name
|
|
}
|
|
|
|
// Process label.selector
|
|
// Always add application metadata label requirements
|
|
appKindReq, err := labels.NewRequirement(ApplicationKindLabel, selection.Equals, []string{r.kindName})
|
|
if err != nil {
|
|
klog.Errorf("Error creating application kind label requirement: %v", err)
|
|
return nil, fmt.Errorf("error creating application kind label requirement: %v", err)
|
|
}
|
|
appGroupReq, err := labels.NewRequirement(ApplicationGroupLabel, selection.Equals, []string{r.gvk.Group})
|
|
if err != nil {
|
|
klog.Errorf("Error creating application group label requirement: %v", err)
|
|
return nil, fmt.Errorf("error creating application group label requirement: %v", err)
|
|
}
|
|
labelRequirements := []labels.Requirement{*appKindReq, *appGroupReq}
|
|
|
|
if options.LabelSelector != nil {
|
|
ls := options.LabelSelector.String()
|
|
parsedLabels, err := labels.Parse(ls)
|
|
if err != nil {
|
|
klog.Errorf("Invalid label selector: %v", err)
|
|
return nil, fmt.Errorf("invalid label selector: %v", err)
|
|
}
|
|
if !parsedLabels.Empty() {
|
|
reqs, _ := parsedLabels.Requirements()
|
|
var prefixedReqs []labels.Requirement
|
|
for _, req := range reqs {
|
|
// Add prefix to each label key
|
|
prefixedReq, err := labels.NewRequirement(LabelPrefix+req.Key(), req.Operator(), req.Values().List())
|
|
if err != nil {
|
|
klog.Errorf("Error prefixing label key: %v", err)
|
|
return nil, fmt.Errorf("error prefixing label key: %v", err)
|
|
}
|
|
prefixedReqs = append(prefixedReqs, *prefixedReq)
|
|
}
|
|
labelRequirements = append(labelRequirements, prefixedReqs...)
|
|
}
|
|
}
|
|
helmLabelSelector = labels.NewSelector().Add(labelRequirements...)
|
|
|
|
// Honor SendInitialEvents (WatchList): the client expects all existing
|
|
// objects as ADDED events, then a Bookmark annotated with
|
|
// metav1.InitialEventsAnnotationKey. controller-runtime's cache already
|
|
// replays the ADDED events, so we just emit the terminating bookmark.
|
|
sendInitialEvents := options.SendInitialEvents != nil && *options.SendInitialEvents
|
|
bookmarker := registry.NewInitialEventsBookmarker(sendInitialEvents, options.ResourceVersion, func() runtime.Object {
|
|
app := &appsv1alpha1.Application{}
|
|
app.TypeMeta = metav1.TypeMeta{
|
|
APIVersion: appsv1alpha1.SchemeGroupVersion.String(),
|
|
Kind: r.kindName,
|
|
}
|
|
return app
|
|
})
|
|
|
|
// Create a custom watcher to transform events
|
|
customW := &customWatcher{
|
|
resultChan: make(chan watch.Event),
|
|
stopChan: make(chan struct{}),
|
|
}
|
|
|
|
// Start watch on HelmRelease with label selector only
|
|
// Field selectors are not supported by controller-runtime cache
|
|
// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
|
|
hrList := &helmv2.HelmReleaseList{}
|
|
helmWatcher, err := r.w.Watch(ctx, hrList, &client.ListOptions{
|
|
Namespace: namespace,
|
|
LabelSelector: helmLabelSelector,
|
|
// Ask the backing watch for bookmarks on a WatchList request; the
|
|
// apiserver omits them by default, leaving the terminating
|
|
// initial-events-end bookmark with no reliable trigger.
|
|
Raw: &metav1.ListOptions{AllowWatchBookmarks: sendInitialEvents},
|
|
})
|
|
if err != nil {
|
|
klog.Errorf("Error setting up watch for HelmReleases: %v", err)
|
|
return nil, err
|
|
}
|
|
customW.underlying = helmWatcher
|
|
|
|
// Start watch on WorkloadMonitor to detect pod readiness changes.
|
|
// For Tenant applications the WorkloadMonitor lives in a computed child
|
|
// namespace (see computeTenantNamespace), not in the HelmRelease namespace,
|
|
// so scoping the watch to `namespace` would miss all events. Use a
|
|
// cluster-wide watch in that case — label selectors still restrict the
|
|
// stream to the relevant kind/group.
|
|
wmLabelSelector := labels.NewSelector().Add(*appKindReq, *appGroupReq)
|
|
wmList := &cozyv1alpha1.WorkloadMonitorList{}
|
|
wmListOpts := &client.ListOptions{LabelSelector: wmLabelSelector}
|
|
if r.kindName != "Tenant" {
|
|
wmListOpts.Namespace = namespace
|
|
}
|
|
wmWatcher, err := r.w.Watch(ctx, wmList, wmListOpts)
|
|
if err != nil {
|
|
klog.Warningf("Failed to set up WorkloadMonitor watch, workload status changes won't trigger events: %v", err)
|
|
// Non-fatal: proceed without WorkloadMonitor watch
|
|
wmWatcher = nil
|
|
}
|
|
|
|
go func() {
|
|
// Capture wmWatcher for cleanup; the variable may be set to nil
|
|
// inside the loop when the channel closes, so defer must use this copy.
|
|
wmWatcherForCleanup := wmWatcher
|
|
defer close(customW.resultChan)
|
|
defer customW.underlying.Stop()
|
|
if wmWatcherForCleanup != nil {
|
|
defer wmWatcherForCleanup.Stop()
|
|
}
|
|
|
|
// Buffer of WorkloadMonitor events that arrived before the initial
|
|
// snapshot finished. The watch-list contract requires the stream to
|
|
// deliver all ADDED events followed by the initial-events-end bookmark
|
|
// before any live updates, so we hold WM-triggered Modified events and
|
|
// replay them once the bookmark has been emitted. Without this, a
|
|
// workload whose status flips during the snapshot window (after the
|
|
// Application ADDED but before the bookmark) and then stops changing
|
|
// would leave the client with a stale WorkloadsReady forever.
|
|
var pendingWMEvents []watch.Event
|
|
|
|
// Get the starting resourceVersion from options
|
|
// If client provides resourceVersion (e.g., from a previous List), we should skip
|
|
// objects with resourceVersion <= startingRV (client already has them)
|
|
var startingRV uint64
|
|
if options.ResourceVersion != "" {
|
|
if rv, err := strconv.ParseUint(options.ResourceVersion, 10, 64); err == nil {
|
|
startingRV = rv
|
|
}
|
|
}
|
|
|
|
// send forwards an event, returning false if the watch or context ended.
|
|
send := func(ev watch.Event) bool {
|
|
select {
|
|
case customW.resultChan <- ev:
|
|
return true
|
|
case <-customW.stopChan:
|
|
return false
|
|
case <-ctx.Done():
|
|
return false
|
|
}
|
|
}
|
|
|
|
drainPendingWMEvents := func() bool {
|
|
for _, ev := range pendingWMEvents {
|
|
if !send(ev) {
|
|
return false
|
|
}
|
|
}
|
|
pendingWMEvents = nil
|
|
return true
|
|
}
|
|
|
|
// Process watch events
|
|
for {
|
|
select {
|
|
case event, ok := <-customW.underlying.ResultChan():
|
|
if !ok {
|
|
// The watcher has been closed
|
|
klog.Warning("HelmRelease watcher closed")
|
|
// Send initial-events-end bookmark before closing if not yet sent
|
|
if bookmark, ok := bookmarker.OnClose(); ok {
|
|
if send(bookmark) {
|
|
drainPendingWMEvents()
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
// Handle bookmark events - these are critical for informer sync
|
|
if event.Type == watch.Bookmark {
|
|
if hr, ok := event.Object.(*helmv2.HelmRelease); ok {
|
|
// controller-runtime bookmarks after the initial list
|
|
// replay; the first one becomes the terminating marker.
|
|
bookmark, initialEventsEnd := bookmarker.OnBackingBookmark(hr.GetResourceVersion())
|
|
if !send(bookmark) {
|
|
return
|
|
}
|
|
if initialEventsEnd && !drainPendingWMEvents() {
|
|
return
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Check if the object is a *v1.Status
|
|
if status, ok := event.Object.(*metav1.Status); ok {
|
|
klog.V(4).Infof("Received Status object in HelmRelease watch: %v", status.Message)
|
|
continue // Skip processing this event
|
|
}
|
|
|
|
// Proceed with processing HelmRelease objects
|
|
hr, ok := event.Object.(*helmv2.HelmRelease)
|
|
if !ok {
|
|
klog.V(4).Infof("Expected HelmRelease object, got %T", event.Object)
|
|
continue
|
|
}
|
|
|
|
// Track resourceVersion for the eventual bookmark
|
|
bookmarker.Observe(hr.GetResourceVersion())
|
|
|
|
// Apply manual field selector filtering (metadata.name and metadata.namespace)
|
|
// controller-runtime cache doesn't support field selectors
|
|
// See: https://github.com/kubernetes-sigs/controller-runtime/issues/612
|
|
if filterByName != "" && hr.Name != filterByName {
|
|
continue
|
|
}
|
|
if !fieldFilter.MatchesNamespace(hr.Namespace) {
|
|
continue
|
|
}
|
|
|
|
// Note: All HelmReleases already match the required labels due to server-side label selector filtering
|
|
// Convert HelmRelease to Application
|
|
app, err := r.ConvertHelmReleaseToApplication(ctx, hr)
|
|
if err != nil {
|
|
klog.Errorf("Error converting HelmRelease to Application: %v", err)
|
|
continue
|
|
}
|
|
|
|
// Apply field.selector by name if specified
|
|
if resourceName != "" && app.Name != resourceName {
|
|
continue
|
|
}
|
|
|
|
// Apply label.selector
|
|
if options.LabelSelector != nil {
|
|
sel, err := labels.Parse(options.LabelSelector.String())
|
|
if err != nil {
|
|
klog.Errorf("Invalid label selector: %v", err)
|
|
continue
|
|
}
|
|
if !sel.Matches(labels.Set(app.Labels)) {
|
|
continue
|
|
}
|
|
}
|
|
|
|
// Emit the terminating bookmark before the first live event, then
|
|
// replay any buffered WorkloadMonitor events.
|
|
if bookmark, ok := bookmarker.BeforeLiveEvent(event.Type); ok {
|
|
if !send(bookmark) {
|
|
return
|
|
}
|
|
if !drainPendingWMEvents() {
|
|
return
|
|
}
|
|
}
|
|
|
|
// Skip ADDED events based on resourceVersion comparison
|
|
if event.Type == watch.Added && startingRV > 0 {
|
|
objRV, parseErr := strconv.ParseUint(app.ResourceVersion, 10, 64)
|
|
// Skip objects client already has (objRV <= startingRV)
|
|
if parseErr == nil && objRV <= startingRV {
|
|
klog.V(6).Infof("Skipping ADDED event for %s/%s (objRV=%d <= startingRV=%d)",
|
|
app.Namespace, app.Name, objRV, startingRV)
|
|
continue
|
|
}
|
|
}
|
|
// When startingRV == 0, always send ADDED events (client wants full state)
|
|
|
|
// Send event to custom watcher
|
|
if !send(watch.Event{Type: event.Type, Object: &app}) {
|
|
return
|
|
}
|
|
|
|
case wmEvent, ok := <-wmResultChan(wmWatcher):
|
|
if !ok {
|
|
klog.V(4).Info("WorkloadMonitor watcher closed")
|
|
wmWatcher = nil
|
|
continue
|
|
}
|
|
if wmEvent.Type == watch.Bookmark || wmEvent.Type == watch.Error {
|
|
if wmEvent.Type == watch.Error {
|
|
klog.V(4).Infof("WorkloadMonitor watch error event: %v", wmEvent.Object)
|
|
}
|
|
continue
|
|
}
|
|
wm, ok := wmEvent.Object.(*cozyv1alpha1.WorkloadMonitor)
|
|
if !ok {
|
|
continue
|
|
}
|
|
// All WM event types (Added/Modified/Deleted) produce a Modified
|
|
// Application event because the Application itself is what changed
|
|
// from the client's perspective.
|
|
wmAppName, hasLabel := wm.Labels[ApplicationNameLabel]
|
|
if !hasLabel {
|
|
continue
|
|
}
|
|
// Filter: skip WorkloadMonitor events for applications not matching
|
|
// the watch scope (single-resource or field-selector filtered watches)
|
|
hrName := r.releaseConfig.Prefix + wmAppName
|
|
if filterByName != "" && hrName != filterByName {
|
|
continue
|
|
}
|
|
if resourceName != "" && wmAppName != resourceName {
|
|
continue
|
|
}
|
|
|
|
// Locate the owning HelmRelease. For most application kinds the
|
|
// WorkloadMonitor and its HelmRelease live in the same namespace,
|
|
// but Tenant workloads live in a child namespace (see
|
|
// computeTenantNamespace) while the HelmRelease remains in the
|
|
// parent/requested namespace — so the WM-to-HR namespace mapping
|
|
// differs.
|
|
hrNS := wm.Namespace
|
|
if r.kindName == "Tenant" {
|
|
hrNS = namespace
|
|
// Filter out WM events whose child namespace does not
|
|
// correspond to the Tenant in our watched namespace, since
|
|
// the cluster-wide WM watch delivers events for all tenants.
|
|
if r.computeTenantNamespace(namespace, wmAppName) != wm.Namespace {
|
|
continue
|
|
}
|
|
if !fieldFilter.MatchesNamespace(hrNS) {
|
|
continue
|
|
}
|
|
} else if !fieldFilter.MatchesNamespace(hrNS) {
|
|
continue
|
|
}
|
|
hr := &helmv2.HelmRelease{}
|
|
if err := r.c.Get(ctx, client.ObjectKey{Namespace: hrNS, Name: hrName}, hr); err != nil {
|
|
klog.V(4).Infof("Cannot find HelmRelease %s/%s for WorkloadMonitor event: %v", hrNS, hrName, err)
|
|
continue
|
|
}
|
|
// Pass the fresh WorkloadMonitor so conversion uses the latest
|
|
// operational status even if the cache (r.c) has not yet
|
|
// observed this watch event.
|
|
app, err := r.ConvertHelmReleaseToApplicationWithMonitor(ctx, hr, wm)
|
|
if err != nil {
|
|
klog.V(4).Infof("Error converting HelmRelease for WorkloadMonitor event: %v", err)
|
|
continue
|
|
}
|
|
// Apply label selector filtering (same as HelmRelease event path)
|
|
if options.LabelSelector != nil {
|
|
sel, err := labels.Parse(options.LabelSelector.String())
|
|
if err != nil {
|
|
klog.Errorf("Invalid label selector: %v", err)
|
|
continue
|
|
}
|
|
if !sel.Matches(labels.Set(app.Labels)) {
|
|
continue
|
|
}
|
|
}
|
|
// Use the WorkloadMonitor's ResourceVersion for the emitted event
|
|
// so clients see a monotonically increasing RV and don't skip this update.
|
|
app.SetResourceVersion(wm.GetResourceVersion())
|
|
outEvent := watch.Event{Type: watch.Modified, Object: &app}
|
|
// Buffer WM-triggered events that arrive before the
|
|
// initial-events-end bookmark. They will be replayed in order
|
|
// immediately after the bookmark is emitted.
|
|
if !bookmarker.Sent() {
|
|
pendingWMEvents = append(pendingWMEvents, outEvent)
|
|
continue
|
|
}
|
|
bookmarker.Observe(wm.GetResourceVersion())
|
|
if !send(outEvent) {
|
|
return
|
|
}
|
|
|
|
case <-customW.stopChan:
|
|
return
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
klog.V(6).Infof("Custom watch established successfully")
|
|
return customW, nil
|
|
}
|
|
|
|
// wmResultChan returns the result channel of a WorkloadMonitor watcher, or a nil
|
|
// channel (which blocks forever in select) if the watcher is nil.
|
|
func wmResultChan(w watch.Interface) <-chan watch.Event {
|
|
if w == nil {
|
|
return nil
|
|
}
|
|
return w.ResultChan()
|
|
}
|
|
|
|
// customWatcher wraps the original watcher and filters/converts events
|
|
type customWatcher struct {
|
|
resultChan chan watch.Event
|
|
stopChan chan struct{}
|
|
stopOnce sync.Once
|
|
underlying watch.Interface
|
|
}
|
|
|
|
// Stop terminates the watch
|
|
func (cw *customWatcher) Stop() {
|
|
cw.stopOnce.Do(func() {
|
|
close(cw.stopChan)
|
|
if cw.underlying != nil {
|
|
cw.underlying.Stop()
|
|
}
|
|
})
|
|
}
|
|
|
|
// ResultChan returns the event channel
|
|
func (cw *customWatcher) ResultChan() <-chan watch.Event {
|
|
return cw.resultChan
|
|
}
|
|
|
|
// getNamespace extracts the namespace from the context
|
|
func (r *REST) getNamespace(ctx context.Context) (string, error) {
|
|
namespace, ok := request.NamespaceFrom(ctx)
|
|
if !ok {
|
|
err := fmt.Errorf("namespace not found in context")
|
|
klog.Error(err)
|
|
return "", err
|
|
}
|
|
return namespace, nil
|
|
}
|
|
|
|
// hasRequiredApplicationLabels checks if a HelmRelease has the required application labels
|
|
// matching the REST instance's kind and group
|
|
func (r *REST) hasRequiredApplicationLabels(hr *helmv2.HelmRelease) bool {
|
|
if hr.Labels == nil {
|
|
return false
|
|
}
|
|
return hr.Labels[ApplicationKindLabel] == r.kindName &&
|
|
hr.Labels[ApplicationGroupLabel] == r.gvk.Group
|
|
}
|
|
|
|
// hasRequiredApplicationLabelsWithName checks if a HelmRelease has the required application labels
|
|
// matching the REST instance's kind, group, and the specified application name
|
|
func (r *REST) hasRequiredApplicationLabelsWithName(hr *helmv2.HelmRelease, appName string) bool {
|
|
if hr.Labels == nil {
|
|
return false
|
|
}
|
|
return hr.Labels[ApplicationKindLabel] == r.kindName &&
|
|
hr.Labels[ApplicationGroupLabel] == r.gvk.Group &&
|
|
hr.Labels[ApplicationNameLabel] == appName
|
|
}
|
|
|
|
// mergeMaps combines two maps of labels or annotations
|
|
func mergeMaps(a, b map[string]string) map[string]string {
|
|
if a == nil && b == nil {
|
|
return nil
|
|
}
|
|
if a == nil {
|
|
return b
|
|
}
|
|
if b == nil {
|
|
return a
|
|
}
|
|
merged := make(map[string]string, len(a)+len(b))
|
|
for k, v := range a {
|
|
merged[k] = v
|
|
}
|
|
for k, v := range b {
|
|
merged[k] = v
|
|
}
|
|
return merged
|
|
}
|
|
|
|
// addPrefixedMap adds the predefined prefix to the keys of a map
|
|
func addPrefixedMap(original map[string]string, prefix string) map[string]string {
|
|
if original == nil {
|
|
return nil
|
|
}
|
|
processed := make(map[string]string, len(original))
|
|
for k, v := range original {
|
|
processed[prefix+k] = v
|
|
}
|
|
return processed
|
|
}
|
|
|
|
// filterPrefixedMap filters a map by the predefined prefix and removes the prefix from the keys
|
|
func filterPrefixedMap(original map[string]string, prefix string) map[string]string {
|
|
if original == nil {
|
|
return nil
|
|
}
|
|
processed := make(map[string]string)
|
|
for k, v := range original {
|
|
if strings.HasPrefix(k, prefix) {
|
|
newKey := strings.TrimPrefix(k, prefix)
|
|
processed[newKey] = v
|
|
}
|
|
}
|
|
return processed
|
|
}
|
|
|
|
// ConvertHelmReleaseToApplication converts a HelmRelease to an Application.
|
|
func (r *REST) ConvertHelmReleaseToApplication(ctx context.Context, hr *helmv2.HelmRelease) (appsv1alpha1.Application, error) {
|
|
return r.ConvertHelmReleaseToApplicationWithMonitor(ctx, hr, nil)
|
|
}
|
|
|
|
// ConvertHelmReleaseToApplicationWithMonitor converts a HelmRelease to an
|
|
// Application, optionally overriding the cached copy of a WorkloadMonitor with
|
|
// a fresher version received from the watch client. This prevents the emitted
|
|
// Application object from carrying stale WorkloadsReady data when r.c (cache)
|
|
// lags behind r.w (watch).
|
|
func (r *REST) ConvertHelmReleaseToApplicationWithMonitor(ctx context.Context, hr *helmv2.HelmRelease, freshMonitor *cozyv1alpha1.WorkloadMonitor) (appsv1alpha1.Application, error) {
|
|
klog.V(6).Infof("Converting HelmRelease to Application for resource %s", hr.GetName())
|
|
|
|
// Convert HelmRelease struct to Application struct
|
|
app, err := r.convertHelmReleaseToApplication(ctx, hr, freshMonitor)
|
|
if err != nil {
|
|
klog.Errorf("Error converting from HelmRelease to Application: %v", err)
|
|
return appsv1alpha1.Application{}, err
|
|
}
|
|
|
|
if err := r.applySpecDefaults(&app); err != nil {
|
|
return app, fmt.Errorf("defaulting error: %w", err)
|
|
}
|
|
|
|
klog.V(6).Infof("Successfully converted HelmRelease %s to Application", hr.GetName())
|
|
return app, nil
|
|
}
|
|
|
|
// ConvertApplicationToHelmRelease converts an Application to a HelmRelease
|
|
func (r *REST) ConvertApplicationToHelmRelease(app *appsv1alpha1.Application) (*helmv2.HelmRelease, error) {
|
|
return r.convertApplicationToHelmRelease(app)
|
|
}
|
|
|
|
// filterInternalKeys removes keys starting with "_" from the JSON values
|
|
func filterInternalKeys(values *apiextv1.JSON) *apiextv1.JSON {
|
|
if values == nil || len(values.Raw) == 0 {
|
|
return values
|
|
}
|
|
var data map[string]any
|
|
if err := json.Unmarshal(values.Raw, &data); err != nil {
|
|
return values
|
|
}
|
|
for key := range data {
|
|
if strings.HasPrefix(key, "_") {
|
|
delete(data, key)
|
|
}
|
|
}
|
|
filtered, err := json.Marshal(data)
|
|
if err != nil {
|
|
return values
|
|
}
|
|
return &apiextv1.JSON{Raw: filtered}
|
|
}
|
|
|
|
// validateNoInternalKeys checks that values don't contain keys starting with "_"
|
|
func validateNoInternalKeys(values *apiextv1.JSON) error {
|
|
if values == nil || len(values.Raw) == 0 {
|
|
return nil
|
|
}
|
|
var data map[string]any
|
|
if err := json.Unmarshal(values.Raw, &data); err != nil {
|
|
return err
|
|
}
|
|
for key := range data {
|
|
if strings.HasPrefix(key, "_") {
|
|
return fmt.Errorf("values key %q is reserved (keys starting with '_' are not allowed)", key)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// maxHelmReleaseName is the Helm release name limit. Helm reserves room for
|
|
// chart-generated resource suffixes within the 63-char DNS-1035 label limit.
|
|
const maxHelmReleaseName = 53
|
|
|
|
// maxNamespaceName is the DNS-1123 label limit for Kubernetes namespace names.
|
|
// The tenant Helm chart creates a Namespace whose name is the computed
|
|
// workload namespace (parent namespace + "-" + tenant name), so the total
|
|
// must fit inside a single 63-char DNS-1123 label.
|
|
const maxNamespaceName = 63
|
|
|
|
// validateNameFormat checks an Application name against DNS-1035 and any
|
|
// kind-specific format rules (e.g. Tenant names must be alphanumeric — see
|
|
// validation.ValidateApplicationName for the reasoning).
|
|
func (r *REST) validateNameFormat(name string) field.ErrorList {
|
|
return validation.ValidateApplicationName(name, r.kindName, field.NewPath("metadata").Child("name"))
|
|
}
|
|
|
|
// validateNameLength checks that the application name won't exceed Kubernetes limits.
|
|
// prefix + name must fit within the Helm release name limit (53 chars).
|
|
func (r *REST) validateNameLength(name string) field.ErrorList {
|
|
fldPath := field.NewPath("metadata").Child("name")
|
|
allErrs := field.ErrorList{}
|
|
|
|
maxLen := maxHelmReleaseName - len(r.releaseConfig.Prefix)
|
|
|
|
if maxLen <= 0 {
|
|
allErrs = append(allErrs, field.Invalid(fldPath, name,
|
|
fmt.Sprintf("configuration error: no valid name length possible (release prefix %q)", r.releaseConfig.Prefix)))
|
|
return allErrs
|
|
}
|
|
|
|
if len(name) > maxLen {
|
|
allErrs = append(allErrs, field.Invalid(fldPath, name,
|
|
fmt.Sprintf("must be no more than %d characters (release prefix %q)", maxLen, r.releaseConfig.Prefix)))
|
|
}
|
|
return allErrs
|
|
}
|
|
|
|
// validateTenantNamespaceLength checks that the computed workload namespace
|
|
// for a Tenant application fits within the DNS-1123 label limit. The namespace
|
|
// is formed by dash-joining the parent namespace with the tenant name, so
|
|
// deep nesting can exceed the limit even when each individual name passes the
|
|
// per-name Helm release length check.
|
|
func (r *REST) validateTenantNamespaceLength(currentNamespace, tenantName string) field.ErrorList {
|
|
fldPath := field.NewPath("metadata").Child("name")
|
|
allErrs := field.ErrorList{}
|
|
|
|
computed := r.computeTenantNamespace(currentNamespace, tenantName)
|
|
if len(computed) > maxNamespaceName {
|
|
allErrs = append(allErrs, field.Invalid(fldPath, tenantName,
|
|
fmt.Sprintf("computed tenant namespace %q would be %d characters, which exceeds the %d-character Kubernetes namespace limit; shorten the tenant name or reduce ancestor nesting depth",
|
|
computed, len(computed), maxNamespaceName)))
|
|
}
|
|
return allErrs
|
|
}
|
|
|
|
// convertHelmReleaseToApplication implements the actual conversion logic.
|
|
// The optional freshMonitor is used to override the cache copy of a
|
|
// WorkloadMonitor when a newer version was delivered via the watch client —
|
|
// see ConvertHelmReleaseToApplicationWithMonitor for the rationale.
|
|
func (r *REST) convertHelmReleaseToApplication(ctx context.Context, hr *helmv2.HelmRelease, freshMonitor *cozyv1alpha1.WorkloadMonitor) (appsv1alpha1.Application, error) {
|
|
// Filter out internal keys (starting with "_") from spec
|
|
filteredSpec := filterInternalKeys(hr.Spec.Values)
|
|
|
|
app := appsv1alpha1.Application{
|
|
TypeMeta: metav1.TypeMeta{
|
|
APIVersion: "apps.cozystack.io/v1alpha1",
|
|
Kind: r.kindName,
|
|
},
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: strings.TrimPrefix(hr.Name, r.releaseConfig.Prefix),
|
|
Namespace: hr.Namespace,
|
|
UID: hr.GetUID(),
|
|
ResourceVersion: hr.GetResourceVersion(),
|
|
CreationTimestamp: hr.CreationTimestamp,
|
|
DeletionTimestamp: hr.DeletionTimestamp,
|
|
Labels: filterPrefixedMap(hr.Labels, LabelPrefix),
|
|
Annotations: filterPrefixedMap(hr.Annotations, AnnotationPrefix),
|
|
},
|
|
Spec: filteredSpec,
|
|
Status: appsv1alpha1.ApplicationStatus{
|
|
Version: hr.Status.LastAttemptedRevision,
|
|
},
|
|
}
|
|
|
|
var conditions []metav1.Condition
|
|
for _, hrCondition := range hr.GetConditions() {
|
|
if hrCondition.Type == "Ready" || hrCondition.Type == "Released" {
|
|
conditions = append(conditions, metav1.Condition{
|
|
LastTransitionTime: hrCondition.LastTransitionTime,
|
|
Reason: hrCondition.Reason,
|
|
Message: hrCondition.Message,
|
|
Status: hrCondition.Status,
|
|
Type: hrCondition.Type,
|
|
})
|
|
}
|
|
}
|
|
// Enrich conditions with WorkloadMonitor operational status.
|
|
// Tenant workloads live in a child namespace (computed from the Tenant name),
|
|
// not in the same namespace as the owning HelmRelease — look there instead.
|
|
workloadsNS := hr.Namespace
|
|
if r.kindName == "Tenant" {
|
|
workloadsNS = r.computeTenantNamespace(hr.Namespace, app.Name)
|
|
}
|
|
ws, wsErr := r.getWorkloadsOperational(ctx, workloadsNS, app.Name, freshMonitor)
|
|
// Derive a stable LastTransitionTime: use the owning HelmRelease's own
|
|
// condition update time (or CreationTimestamp as a floor) so that repeated
|
|
// conversions of the same underlying state produce identical timestamps.
|
|
wrTransition := hr.CreationTimestamp
|
|
for _, c := range hr.GetConditions() {
|
|
if c.LastTransitionTime.After(wrTransition.Time) {
|
|
wrTransition = c.LastTransitionTime
|
|
}
|
|
}
|
|
if ws.transitionTime.After(wrTransition.Time) {
|
|
wrTransition = ws.transitionTime
|
|
}
|
|
if wrTransition.IsZero() {
|
|
// Fallback for objects that somehow have no timestamps at all
|
|
// (e.g. hand-crafted test fixtures). In production HelmReleases
|
|
// always carry a CreationTimestamp, so the stable branch above
|
|
// is used.
|
|
wrTransition = metav1.Now()
|
|
}
|
|
if wsErr != nil {
|
|
// Fail-open: if we can't query WorkloadMonitors (e.g., informer cache not ready),
|
|
// don't override Ready. Prefer operational availability over safety.
|
|
// The WorkloadsReady=Unknown condition still signals the issue to the user.
|
|
klog.Warningf("Failed to check workload monitors for %s/%s: %v", workloadsNS, app.Name, wsErr)
|
|
conditions = append(conditions, metav1.Condition{
|
|
Type: "WorkloadsReady",
|
|
Status: metav1.ConditionUnknown,
|
|
LastTransitionTime: wrTransition,
|
|
Reason: "Error",
|
|
Message: fmt.Sprintf("Failed to check workload status: %v", wsErr),
|
|
})
|
|
} else if ws.found {
|
|
workloadsCondition := metav1.Condition{
|
|
Type: "WorkloadsReady",
|
|
LastTransitionTime: wrTransition,
|
|
Reason: "WorkloadMonitorCheck",
|
|
}
|
|
switch {
|
|
case !ws.operational:
|
|
// Concrete failure takes priority over unknown/pending state
|
|
workloadsCondition.Status = metav1.ConditionFalse
|
|
workloadsCondition.Message = "One or more workloads are not operational"
|
|
case ws.unknown:
|
|
workloadsCondition.Status = metav1.ConditionUnknown
|
|
workloadsCondition.Reason = "Pending"
|
|
workloadsCondition.Message = "One or more workloads have not been reconciled yet"
|
|
default:
|
|
workloadsCondition.Status = metav1.ConditionTrue
|
|
workloadsCondition.Message = "All workloads are operational"
|
|
}
|
|
conditions = append(conditions, workloadsCondition)
|
|
|
|
// Intentionally do NOT override the Ready condition based on WorkloadsReady.
|
|
// Ready continues to reflect HelmRelease state only, which:
|
|
// - preserves backward compatibility with existing tooling (kubectl wait,
|
|
// GitOps health checks) that expect Ready to match HelmRelease
|
|
// - avoids false-negative Ready=False during normal startup windows where
|
|
// pods are still coming up but WorkloadMonitor has already reported
|
|
// Operational=false due to availableReplicas < MinReplicas
|
|
// WorkloadsReady is a separate condition that surfaces workload health
|
|
// independently — users and dashboards can observe it for operational visibility.
|
|
}
|
|
|
|
app.SetConditions(conditions)
|
|
|
|
// Add namespace field for Tenant applications
|
|
if r.kindName == validation.TenantKind {
|
|
app.Status.Namespace = r.computeTenantNamespace(hr.Namespace, app.Name)
|
|
externalIPsCount, err := r.countTenantExternalIPs(ctx, app.Status.Namespace)
|
|
if err != nil {
|
|
klog.Warningf("Failed to count external IPs for tenant %s/%s: %v", hr.Namespace, app.Name, err)
|
|
} else {
|
|
app.Status.ExternalIPsCount = externalIPsCount
|
|
}
|
|
}
|
|
|
|
return app, nil
|
|
}
|
|
|
|
// workloadsStatus holds the aggregated operational status of WorkloadMonitors.
|
|
type workloadsStatus struct {
|
|
operational bool
|
|
found bool
|
|
unknown bool // true when at least one monitor has nil Operational (not yet reconciled)
|
|
// transitionTime is the most recent metadata update time across the
|
|
// matching monitors. Used as WorkloadsReady.LastTransitionTime so that
|
|
// repeated conversions for the same underlying state produce stable
|
|
// timestamps (preserving the Kubernetes contract that identical
|
|
// resource versions represent identical content).
|
|
transitionTime metav1.Time
|
|
}
|
|
|
|
// getWorkloadsOperational checks WorkloadMonitor resources for an application and returns
|
|
// aggregated operational status. If no monitors exist, returns found=false.
|
|
// When freshOverride is non-nil, its status replaces the cached copy for the
|
|
// corresponding monitor — this keeps the result consistent with watch events
|
|
// when the cache (r.c) lags behind the watch client (r.w).
|
|
func (r *REST) getWorkloadsOperational(ctx context.Context, namespace, appName string, freshOverride *cozyv1alpha1.WorkloadMonitor) (workloadsStatus, error) {
|
|
monitors := &cozyv1alpha1.WorkloadMonitorList{}
|
|
if err := r.c.List(ctx, monitors,
|
|
client.InNamespace(namespace),
|
|
client.MatchingLabels{
|
|
appsv1alpha1.ApplicationKindLabel: r.kindName,
|
|
appsv1alpha1.ApplicationGroupLabel: r.gvk.Group,
|
|
appsv1alpha1.ApplicationNameLabel: appName,
|
|
},
|
|
); err != nil {
|
|
return workloadsStatus{}, err
|
|
}
|
|
// Ensure the freshOverride is represented in the aggregation even when
|
|
// the cache has not yet observed it (brand-new resource) or is behind.
|
|
replaced := false
|
|
if freshOverride != nil {
|
|
for i := range monitors.Items {
|
|
if monitors.Items[i].UID == freshOverride.UID ||
|
|
(monitors.Items[i].Name == freshOverride.Name && monitors.Items[i].Namespace == freshOverride.Namespace) {
|
|
monitors.Items[i] = *freshOverride
|
|
replaced = true
|
|
break
|
|
}
|
|
}
|
|
if !replaced {
|
|
monitors.Items = append(monitors.Items, *freshOverride)
|
|
}
|
|
}
|
|
if len(monitors.Items) == 0 {
|
|
return workloadsStatus{operational: true, found: false}, nil
|
|
}
|
|
operational := true
|
|
unknown := false
|
|
var latest metav1.Time
|
|
for _, m := range monitors.Items {
|
|
if m.Status.Operational == nil {
|
|
unknown = true
|
|
} else if !*m.Status.Operational {
|
|
operational = false
|
|
}
|
|
// Pick the most recent monitor mtime as a stable transition time.
|
|
if t := latestMonitorTime(&m); t.After(latest.Time) {
|
|
latest = t
|
|
}
|
|
}
|
|
return workloadsStatus{operational: operational, found: true, unknown: unknown, transitionTime: latest}, nil
|
|
}
|
|
|
|
// latestMonitorTime returns the most recent timestamp associated with a
|
|
// WorkloadMonitor — currently only the object creation time is guaranteed.
|
|
// Status does not carry a transition time, so we fall back to CreationTimestamp.
|
|
func latestMonitorTime(m *cozyv1alpha1.WorkloadMonitor) metav1.Time {
|
|
return m.CreationTimestamp
|
|
}
|
|
|
|
// convertApplicationToHelmRelease implements the actual conversion logic.
|
|
//
|
|
// Spec.Interval, Install/Upgrade.Strategy{Name=RetryOnFailure,RetryInterval},
|
|
// Install/Upgrade.Timeout, and Spec.MaxHistory are populated from
|
|
// ReleaseConfig fields fed by the cozystack-api server flags
|
|
// (--helmrelease-{interval,retry-interval,install-timeout,upgrade-timeout,
|
|
// max-history}). This mirrors cozystack-operator's
|
|
// PackageReconciler.buildHelmReleaseSpec so both HelmRelease-generating paths
|
|
// share the same retry strategy, history retention, and reconcile cadence.
|
|
//
|
|
// Strategy.Name=RetryOnFailure with an explicit RetryInterval (rather than
|
|
// Remediation{Retries:-1}) decouples failed install/upgrade retry timing
|
|
// from Spec.Interval — the previous coupling caused failed installs to
|
|
// retry only every 5m, exceeding E2E budgets.
|
|
func (r *REST) convertApplicationToHelmRelease(app *appsv1alpha1.Application) (*helmv2.HelmRelease, error) {
|
|
// Per-Application annotation overrides win over the global defaults
|
|
// (HelmReleaseInstallTimeout / HelmReleaseUpgradeTimeout):
|
|
// - HelmInstallTimeout (release.cozystack.io/helm-install-timeout)
|
|
// sets both Install.Timeout and Upgrade.Timeout;
|
|
// - HelmUpgradeTimeout (release.cozystack.io/helm-upgrade-timeout)
|
|
// then overrides only Upgrade.Timeout, so a kind can carry an
|
|
// asymmetric budget (short install, long upgrade or vice versa).
|
|
// kubernetes-rd and tenant-rd carry helm-install-timeout today: the
|
|
// Kubernetes Application's parent chart contains CAPI/Kamaji
|
|
// resources whose admin-kubeconfig Secret is provisioned
|
|
// asynchronously and Kamaji cold-start routinely exceeds flux's
|
|
// default wait budget, and the Tenant parent chart bootstraps the
|
|
// seaweedfs-db CNPG cluster whose first reconcile exceeds it too.
|
|
// Any future kind with the same shape can opt in by setting the
|
|
// same annotation.
|
|
installTimeout := r.releaseConfig.HelmReleaseInstallTimeout
|
|
upgradeTimeout := r.releaseConfig.HelmReleaseUpgradeTimeout
|
|
if r.releaseConfig.HelmInstallTimeout > 0 {
|
|
installTimeout = r.releaseConfig.HelmInstallTimeout
|
|
upgradeTimeout = r.releaseConfig.HelmInstallTimeout
|
|
}
|
|
if r.releaseConfig.HelmUpgradeTimeout > 0 {
|
|
upgradeTimeout = r.releaseConfig.HelmUpgradeTimeout
|
|
}
|
|
|
|
maxHistory := r.releaseConfig.HelmReleaseMaxHistory
|
|
helmRelease := &helmv2.HelmRelease{
|
|
TypeMeta: metav1.TypeMeta{
|
|
APIVersion: "helm.toolkit.fluxcd.io/v2",
|
|
Kind: "HelmRelease",
|
|
},
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: r.releaseConfig.Prefix + app.Name,
|
|
Namespace: app.Namespace,
|
|
Labels: addPrefixedMap(app.Labels, LabelPrefix),
|
|
Annotations: addPrefixedMap(app.Annotations, AnnotationPrefix),
|
|
ResourceVersion: app.ResourceVersion,
|
|
UID: app.UID,
|
|
},
|
|
Spec: helmv2.HelmReleaseSpec{
|
|
ChartRef: &helmv2.CrossNamespaceSourceReference{
|
|
Kind: r.releaseConfig.ChartRef.Kind,
|
|
Name: r.releaseConfig.ChartRef.Name,
|
|
Namespace: r.releaseConfig.ChartRef.Namespace,
|
|
},
|
|
Interval: metav1.Duration{Duration: r.releaseConfig.HelmReleaseInterval},
|
|
MaxHistory: &maxHistory,
|
|
Install: &helmv2.Install{
|
|
Timeout: &metav1.Duration{Duration: installTimeout},
|
|
Strategy: &helmv2.InstallStrategy{
|
|
Name: string(helmv2.ActionStrategyRetryOnFailure),
|
|
RetryInterval: &metav1.Duration{Duration: r.releaseConfig.HelmReleaseRetryInterval},
|
|
},
|
|
},
|
|
Upgrade: &helmv2.Upgrade{
|
|
Timeout: &metav1.Duration{Duration: upgradeTimeout},
|
|
Strategy: &helmv2.UpgradeStrategy{
|
|
Name: string(helmv2.ActionStrategyRetryOnFailure),
|
|
RetryInterval: &metav1.Duration{Duration: r.releaseConfig.HelmReleaseRetryInterval},
|
|
},
|
|
},
|
|
ValuesFrom: []helmv2.ValuesReference{
|
|
{
|
|
Kind: "Secret",
|
|
Name: "cozystack-values",
|
|
},
|
|
},
|
|
Values: app.Spec,
|
|
},
|
|
}
|
|
|
|
// Per-Application HelmRelease wait disablement. An
|
|
// ApplicationDefinition that sets
|
|
// release.cozystack.io/helm-install-disable-wait: "true" gets
|
|
// Install.DisableWait and Upgrade.DisableWait set on the emitted
|
|
// HelmRelease. Required for parent charts that emit in-tenant child
|
|
// HelmReleases which cannot become Ready during the parent's own
|
|
// install (e.g. the Kubernetes Application emits cilium/coredns/csi/etc.
|
|
// HelmReleases that can never become Ready until worker nodes exist,
|
|
// plus a main-phase talos-reconcile Job that generates the
|
|
// TalosConfigTemplate the worker MachineSet clones from; without
|
|
// DisableWait the helm-controller blocks the release on those addon
|
|
// HelmReleases, which cannot happen during install).
|
|
if r.releaseConfig.HelmInstallDisableWait {
|
|
helmRelease.Spec.Install.DisableWait = true
|
|
helmRelease.Spec.Upgrade.DisableWait = true
|
|
}
|
|
|
|
// kstatus readiness (issue #2642): set the wait strategy + CEL health
|
|
// expressions so the release reports Ready only when the rendered CR is
|
|
// actually healthy instead of as soon as helm applies it. ResolveWaitStrategy
|
|
// couples the default — expressions imply the poller strategy when none is
|
|
// set, since healthCheckExprs are only evaluated under poller.
|
|
helmRelease.Spec.HealthCheckExprs = r.releaseConfig.HealthCheckExprs
|
|
helmRelease.Spec.WaitStrategy = config.ResolveWaitStrategy(r.releaseConfig.WaitStrategy, len(r.releaseConfig.HealthCheckExprs) > 0)
|
|
|
|
return helmRelease, nil
|
|
}
|
|
|
|
// ConvertToTable implements the TableConvertor interface for displaying resources in a table format
|
|
func (r *REST) ConvertToTable(ctx context.Context, object runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
|
|
klog.V(6).Infof("ConvertToTable: received object of type %T", object)
|
|
|
|
var table metav1.Table
|
|
|
|
switch obj := object.(type) {
|
|
case *appsv1alpha1.ApplicationList:
|
|
table = r.buildTableFromApplications(obj.Items)
|
|
table.ResourceVersion = obj.ResourceVersion
|
|
case *appsv1alpha1.Application:
|
|
table = r.buildTableFromApplication(*obj)
|
|
table.ResourceVersion = obj.GetResourceVersion()
|
|
default:
|
|
resource := schema.GroupResource{}
|
|
if info, ok := request.RequestInfoFrom(ctx); ok {
|
|
resource = schema.GroupResource{Group: info.APIGroup, Resource: info.Resource}
|
|
}
|
|
return nil, errNotAcceptable{
|
|
resource: resource,
|
|
message: "object does not implement the Object interfaces",
|
|
}
|
|
}
|
|
|
|
// Handle table options
|
|
if opt, ok := tableOptions.(*metav1.TableOptions); ok && opt != nil && opt.NoHeaders {
|
|
table.ColumnDefinitions = nil
|
|
}
|
|
|
|
table.TypeMeta = metav1.TypeMeta{
|
|
APIVersion: "meta.k8s.io/v1",
|
|
Kind: "Table",
|
|
}
|
|
|
|
klog.V(6).Infof("ConvertToTable: returning table with %d rows", len(table.Rows))
|
|
return &table, nil
|
|
}
|
|
|
|
// buildTableFromApplications constructs a table from a list of Applications
|
|
func (r *REST) buildTableFromApplications(apps []appsv1alpha1.Application) metav1.Table {
|
|
table := metav1.Table{
|
|
ColumnDefinitions: []metav1.TableColumnDefinition{
|
|
{Name: "NAME", Type: "string", Description: "Name of the Application", Priority: 0},
|
|
{Name: "READY", Type: "string", Description: "Ready status of the Application", Priority: 0},
|
|
{Name: "AGE", Type: "string", Description: "Age of the Application", Priority: 0},
|
|
{Name: "VERSION", Type: "string", Description: "Version of the Application", Priority: 0},
|
|
},
|
|
Rows: make([]metav1.TableRow, 0, len(apps)),
|
|
}
|
|
now := time.Now()
|
|
|
|
for i := range apps {
|
|
app := &apps[i]
|
|
row := metav1.TableRow{
|
|
Cells: []any{app.GetName(), getReadyStatus(app.Status.Conditions), computeAge(app.GetCreationTimestamp().Time, now), getVersion(app.Status.Version)},
|
|
Object: runtime.RawExtension{Object: app},
|
|
}
|
|
table.Rows = append(table.Rows, row)
|
|
}
|
|
|
|
return table
|
|
}
|
|
|
|
// buildTableFromApplication constructs a table from a single Application
|
|
func (r *REST) buildTableFromApplication(app appsv1alpha1.Application) metav1.Table {
|
|
table := metav1.Table{
|
|
ColumnDefinitions: []metav1.TableColumnDefinition{
|
|
{Name: "NAME", Type: "string", Description: "Name of the Application", Priority: 0},
|
|
{Name: "READY", Type: "string", Description: "Ready status of the Application", Priority: 0},
|
|
{Name: "AGE", Type: "string", Description: "Age of the Application", Priority: 0},
|
|
{Name: "VERSION", Type: "string", Description: "Version of the Application", Priority: 0},
|
|
},
|
|
Rows: []metav1.TableRow{},
|
|
}
|
|
now := time.Now()
|
|
|
|
a := app
|
|
row := metav1.TableRow{
|
|
Cells: []any{app.GetName(), getReadyStatus(app.Status.Conditions), computeAge(app.GetCreationTimestamp().Time, now), getVersion(app.Status.Version)},
|
|
Object: runtime.RawExtension{Object: &a},
|
|
}
|
|
table.Rows = append(table.Rows, row)
|
|
|
|
return table
|
|
}
|
|
|
|
// getVersion extracts and returns only the revision from the version string
|
|
// If version is in format "0.1.4+abcdef", returns "abcdef"
|
|
// Otherwise returns the original string or "<unknown>" if empty
|
|
func getVersion(version string) string {
|
|
if version == "" {
|
|
return "<unknown>"
|
|
}
|
|
// Check if version contains "+" separator
|
|
if idx := strings.LastIndex(version, "+"); idx >= 0 && idx < len(version)-1 {
|
|
// Return only the part after "+"
|
|
return version[idx+1:]
|
|
}
|
|
// If no "+" found, return original version
|
|
return version
|
|
}
|
|
|
|
// computeAge calculates the age of the object based on CreationTimestamp and current time
|
|
func computeAge(creationTime, currentTime time.Time) string {
|
|
ageDuration := currentTime.Sub(creationTime)
|
|
return duration.HumanDuration(ageDuration)
|
|
}
|
|
|
|
// getReadyStatus returns the ready status based on conditions
|
|
func getReadyStatus(conditions []metav1.Condition) string {
|
|
for _, condition := range conditions {
|
|
if condition.Type == "Ready" {
|
|
switch condition.Status {
|
|
case metav1.ConditionTrue:
|
|
return "True"
|
|
case metav1.ConditionFalse:
|
|
return "False"
|
|
default:
|
|
return "Unknown"
|
|
}
|
|
}
|
|
}
|
|
return "Unknown"
|
|
}
|
|
|
|
// computeTenantNamespace computes the namespace for a Tenant application based on the specified logic
|
|
func (r *REST) computeTenantNamespace(currentNamespace, tenantName string) string {
|
|
hrName := r.releaseConfig.Prefix + tenantName
|
|
|
|
switch {
|
|
case currentNamespace == "tenant-root" && hrName == "tenant-root":
|
|
// 1) root tenant inside root namespace
|
|
return "tenant-root"
|
|
|
|
case currentNamespace == "tenant-root":
|
|
// 2) any other tenant in root namespace
|
|
return fmt.Sprintf("tenant-%s", tenantName)
|
|
|
|
default:
|
|
// 3) tenant in a dedicated namespace
|
|
return fmt.Sprintf("%s-%s", currentNamespace, tenantName)
|
|
}
|
|
}
|
|
|
|
func (r *REST) countTenantExternalIPs(ctx context.Context, namespace string) (int32, error) {
|
|
if namespace == "" {
|
|
return 0, nil
|
|
}
|
|
|
|
var services corev1.ServiceList
|
|
if err := r.c.List(
|
|
ctx,
|
|
&services,
|
|
client.InNamespace(namespace),
|
|
client.MatchingFields{"spec.type": string(corev1.ServiceTypeLoadBalancer)},
|
|
); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
var count int32
|
|
for i := range services.Items {
|
|
svc := &services.Items[i]
|
|
for _, ingress := range svc.Status.LoadBalancer.Ingress {
|
|
if ingress.IP != "" {
|
|
count++
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
return count, nil
|
|
}
|
|
|
|
// Destroy releases resources associated with REST
|
|
func (r *REST) Destroy() {
|
|
// No additional actions needed to release resources.
|
|
}
|
|
|
|
// New creates a new instance of Application
|
|
func (r *REST) New() runtime.Object {
|
|
obj := &appsv1alpha1.Application{}
|
|
obj.TypeMeta = metav1.TypeMeta{
|
|
APIVersion: r.gvk.GroupVersion().String(),
|
|
Kind: r.kindName,
|
|
}
|
|
return obj
|
|
}
|
|
|
|
// NewList returns an empty list of Application objects
|
|
func (r *REST) NewList() runtime.Object {
|
|
obj := &appsv1alpha1.ApplicationList{}
|
|
obj.TypeMeta = metav1.TypeMeta{
|
|
APIVersion: r.gvk.GroupVersion().String(),
|
|
Kind: r.kindName + "List",
|
|
}
|
|
return obj
|
|
}
|
|
|
|
// Kind returns the resource kind used for API discovery
|
|
func (r *REST) Kind() string {
|
|
return r.gvk.Kind
|
|
}
|
|
|
|
// GroupVersionKind returns the GroupVersionKind for REST
|
|
func (r *REST) GroupVersionKind(schema.GroupVersion) schema.GroupVersionKind {
|
|
return r.gvk
|
|
}
|
|
|
|
// deprecationMessagesFor builds the deprecation log lines for any
|
|
// resourcesPreset field on the Application that still carries a legacy
|
|
// flat name (nano/micro/.../2xlarge). Returned in the order
|
|
// presets.FindLegacyPresets discovers them. Pure function so tests can
|
|
// assert the output without intercepting klog.
|
|
//
|
|
// A malformed spec.Raw is reported via a V(2) debug log instead of a
|
|
// hard error: validation upstream of this hook owns rejecting invalid
|
|
// specs, and surfacing the unmarshal failure as a warning here would
|
|
// fire even on transient inputs the caller is about to reject anyway.
|
|
func deprecationMessagesFor(kindName, namespace, name string, raw []byte) []string {
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
var spec map[string]any
|
|
if err := json.Unmarshal(raw, &spec); err != nil {
|
|
klog.V(2).Infof("deprecationMessagesFor: skipping %s/%s in %s: spec is not valid JSON: %v",
|
|
kindName, name, namespace, err)
|
|
return nil
|
|
}
|
|
return presets.FormatDeprecationMessages(kindName, namespace, name, presets.FindLegacyPresets(spec))
|
|
}
|
|
|
|
// warnLegacyPresets emits a klog warning per deprecation message for
|
|
// the given Application. Non-blocking: the value still renders
|
|
// correctly through cozy-lib legacy aliases.
|
|
func (r *REST) warnLegacyPresets(app *appsv1alpha1.Application) {
|
|
if app == nil || app.Spec == nil {
|
|
return
|
|
}
|
|
for _, msg := range deprecationMessagesFor(r.kindName, app.Namespace, app.Name, app.Spec.Raw) {
|
|
klog.Warning(msg)
|
|
}
|
|
}
|
|
|
|
// errNotAcceptable indicates that the resource does not support conversion to Table
|
|
type errNotAcceptable struct {
|
|
resource schema.GroupResource
|
|
message string
|
|
}
|
|
|
|
func (e errNotAcceptable) Error() string {
|
|
return e.message
|
|
}
|
|
|
|
func (e errNotAcceptable) Status() metav1.Status {
|
|
return metav1.Status{
|
|
Status: metav1.StatusFailure,
|
|
Code: http.StatusNotAcceptable,
|
|
Reason: metav1.StatusReason("NotAcceptable"),
|
|
Message: e.Error(),
|
|
}
|
|
}
|