dualfroz
fd83c655ca
WaitForDeployment returned as soon as AvailableReplicas was high enough, without checking that the status belonged to the current spec. UpdateScale only writes the spec, so a Get issued right after it can still carry the previous generation's status, counting the pod that is being torn down. Gate AvailableReplicas behind ObservedGeneration, the same condition WaitForDeploymentRollout already uses in the integration framework.
215 lines
8.5 KiB
Go
215 lines
8.5 KiB
Go
// SPDX-FileCopyrightText: The Fission Authors
|
||
//
|
||
// SPDX-License-Identifier: Apache-2.0
|
||
|
||
package util
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"strconv"
|
||
"time"
|
||
|
||
"github.com/go-logr/logr"
|
||
appsv1 "k8s.io/api/apps/v1"
|
||
autoscalingv1 "k8s.io/api/autoscaling/v1"
|
||
k8serrors "k8s.io/apimachinery/pkg/api/errors"
|
||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||
"k8s.io/client-go/kubernetes"
|
||
|
||
fv1 "github.com/fission/fission/pkg/apis/core/v1"
|
||
otelUtils "github.com/fission/fission/pkg/utils/otel"
|
||
)
|
||
|
||
// ScaleDeployment scales the named deployment to replicas via the scale
|
||
// subresource. Shared by the newdeploy and container managers, which both
|
||
// scale a per-function deployment up to MinScale on (re)specialization.
|
||
func ScaleDeployment(ctx context.Context, kubeClient kubernetes.Interface, logger logr.Logger, ns, name string, replicas int32) error {
|
||
otelUtils.SpanTrackEvent(ctx, "scaleDeployment", otelUtils.MapToAttributes(map[string]string{
|
||
"deployment-name": name,
|
||
"deployment-namespace": ns,
|
||
"replicas": fmt.Sprintf("%d", replicas),
|
||
})...)
|
||
logger = otelUtils.LoggerWithTraceID(ctx, logger)
|
||
logger.Info("scaling deployment",
|
||
"deployment", name,
|
||
"namespace", ns,
|
||
"replicas", replicas)
|
||
_, err := kubeClient.AppsV1().Deployments(ns).UpdateScale(ctx, name, &autoscalingv1.Scale{
|
||
ObjectMeta: metav1.ObjectMeta{
|
||
Name: name,
|
||
Namespace: ns,
|
||
},
|
||
Spec: autoscalingv1.ScaleSpec{
|
||
Replicas: replicas,
|
||
},
|
||
}, metav1.UpdateOptions{})
|
||
return err
|
||
}
|
||
|
||
// waitPollInitialDelay/waitPollMaxDelay shape WaitForDeployment's adaptive
|
||
// poll: scale-from-zero deployments usually become available well under a
|
||
// second, so polling starts fine-grained and backs off ×1.5 to the old 1s
|
||
// interval — a fixed 1s tick quantized every newdeploy/container cold start
|
||
// up to a full second.
|
||
const (
|
||
waitPollInitialDelay = 50 * time.Millisecond
|
||
waitPollMaxDelay = time.Second
|
||
)
|
||
|
||
// nextWaitPollDelay returns the poll interval following cur.
|
||
func nextWaitPollDelay(cur time.Duration) time.Duration {
|
||
return min(cur*3/2, waitPollMaxDelay)
|
||
}
|
||
|
||
// deploymentAvailable reports whether depl has at least replicas available,
|
||
// as observed by the deployment controller for the spec currently stored.
|
||
//
|
||
// The generation gate matters on the scale-from-zero path. UpdateScale only
|
||
// writes the spec and bumps metadata.generation; status is written later, by
|
||
// the deployment controller. A Get issued right after ScaleDeployment can
|
||
// therefore still carry the status of the previous generation, which counts
|
||
// the pod the idle reaper is tearing down, so AvailableReplicas alone would
|
||
// report a serving deployment while the Service has no endpoint. Until
|
||
// ObservedGeneration catches up, the status says nothing about this scale
|
||
// request. Same gate as WaitForDeploymentRollout in the integration framework.
|
||
func deploymentAvailable(depl *appsv1.Deployment, replicas int32) bool {
|
||
return depl.Status.ObservedGeneration >= depl.Generation && depl.Status.AvailableReplicas >= replicas
|
||
}
|
||
|
||
// WaitForDeployment polls the deployment until the deployment controller has
|
||
// observed the current spec and reports at least replicas AvailableReplicas,
|
||
// or until specializationTimeout seconds elapse. Shared by the newdeploy and
|
||
// container managers. specializationTimeout is floored at
|
||
// fv1.DefaultSpecializationTimeOut. AvailableReplicas is used in preference to
|
||
// ReadyReplicas since the pods may not be able to serve network traffic yet.
|
||
func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, logger logr.Logger, depl *appsv1.Deployment, replicas int32, specializationTimeout int) (latestDepl *appsv1.Deployment, err error) {
|
||
oldStatus := depl.Status
|
||
otelUtils.SpanTrackEvent(ctx, "waitForDeployment", otelUtils.GetAttributesForDeployment(depl)...)
|
||
// if no specializationTimeout is set, use default value
|
||
if specializationTimeout < fv1.DefaultSpecializationTimeOut {
|
||
specializationTimeout = fv1.DefaultSpecializationTimeOut
|
||
}
|
||
|
||
logger = otelUtils.LoggerWithTraceID(ctx, logger)
|
||
|
||
// The timeout is a wall-clock deadline (the poll interval varies, so an
|
||
// iteration count would silently shrink or stretch it).
|
||
deadline := time.Now().Add(time.Duration(specializationTimeout) * time.Second)
|
||
delay := waitPollInitialDelay
|
||
// One reusable timer across polls (Reset is safe on Go ≥1.23 timers)
|
||
// instead of a time.After allocation per iteration.
|
||
timer := time.NewTimer(time.Hour)
|
||
defer timer.Stop()
|
||
|
||
for {
|
||
latestDepl, err = kubeClient.AppsV1().Deployments(depl.ObjectMeta.Namespace).Get(ctx, depl.Name, metav1.GetOptions{})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
// TODO check for imagePullerror
|
||
if deploymentAvailable(latestDepl, replicas) {
|
||
otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...)
|
||
return latestDepl, err
|
||
}
|
||
remaining := time.Until(deadline)
|
||
if remaining <= 0 {
|
||
break
|
||
}
|
||
// Wait before the next poll, but stay responsive to cancellation
|
||
// (executor shutdown / loss of leadership) instead of blocking for the
|
||
// full interval.
|
||
timer.Reset(min(delay, remaining))
|
||
select {
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
case <-timer.C:
|
||
}
|
||
delay = nextWaitPollDelay(delay)
|
||
}
|
||
|
||
logger.Error(nil, "Deployment provision failed within timeout window", "name", latestDepl.Name, "old_status", oldStatus,
|
||
"current_status", latestDepl.Status, "timeout", specializationTimeout)
|
||
|
||
// this error appears in the executor pod logs
|
||
timeoutError := fmt.Errorf("failed to create deployment within the timeout window of %d seconds", specializationTimeout)
|
||
return nil, timeoutError
|
||
}
|
||
|
||
// ReferencedResourcesRVSum returns the sum of the resource versions of the
|
||
// secrets and configmaps the function references. Shared by the newdeploy and
|
||
// container managers.
|
||
//
|
||
// We used to update a timestamp in the deployment environment field in order to
|
||
// trigger a rolling update when the function's referenced resources changed.
|
||
// But a timestamp also changes when the executor merely adopts an orphaned
|
||
// deployment, triggering an unwanted rolling update. The sum of the referenced
|
||
// resources' resource versions reflects content changes without depending on
|
||
// time, so adoption alone does not perturb it.
|
||
//
|
||
// Each reference is fetched by name (rather than listing every secret/configmap
|
||
// in the namespace) to minimise API traffic. Behaviour is preserved: only
|
||
// references resolvable within the deployment namespace contribute, and a
|
||
// missing referenced object contributes nothing rather than erroring.
|
||
//
|
||
// The sum is returned as int64: resource versions are int64 etcd revisions and
|
||
// can exceed 2^31-1 on long-running clusters, so narrowing to int would risk
|
||
// truncation (and collisions that mask a genuine change).
|
||
// The signature takes the whole FunctionSpec because the reference surface is
|
||
// wider than Spec.Secrets/ConfigMaps: objects referenced only via
|
||
// Env[].valueFrom / EnvFrom (RFC-0030) must contribute too — Kubernetes never
|
||
// refreshes env in a running container, so a rotation the sum misses would
|
||
// propagate to nothing (silently stale credentials).
|
||
func ReferencedResourcesRVSum(ctx context.Context, client kubernetes.Interface, namespace string, spec fv1.FunctionSpec) (int64, error) {
|
||
var rvCount int64
|
||
|
||
// Deduplicate: the same object referenced via Spec.Secrets AND an env
|
||
// field must contribute once, or removing one of two references would
|
||
// change the sum and roll pods for a content-identical state.
|
||
secretNames := make(map[string]struct{})
|
||
for _, ref := range spec.Secrets {
|
||
if ref.Namespace == namespace {
|
||
secretNames[ref.Name] = struct{}{}
|
||
}
|
||
}
|
||
// Env references are same-namespace by construction (LocalObjectReference).
|
||
for _, name := range spec.EnvSecretNames() {
|
||
secretNames[name] = struct{}{}
|
||
}
|
||
cfgmapNames := make(map[string]struct{})
|
||
for _, ref := range spec.ConfigMaps {
|
||
if ref.Namespace == namespace {
|
||
cfgmapNames[ref.Name] = struct{}{}
|
||
}
|
||
}
|
||
for _, name := range spec.EnvConfigMapNames() {
|
||
cfgmapNames[name] = struct{}{}
|
||
}
|
||
|
||
for name := range secretNames {
|
||
s, err := client.CoreV1().Secrets(namespace).Get(ctx, name, metav1.GetOptions{})
|
||
if err != nil {
|
||
if k8serrors.IsNotFound(err) {
|
||
continue
|
||
}
|
||
return 0, err
|
||
}
|
||
rv, _ := strconv.ParseInt(s.ResourceVersion, 10, 64)
|
||
rvCount += rv
|
||
}
|
||
|
||
for name := range cfgmapNames {
|
||
cfg, err := client.CoreV1().ConfigMaps(namespace).Get(ctx, name, metav1.GetOptions{})
|
||
if err != nil {
|
||
if k8serrors.IsNotFound(err) {
|
||
continue
|
||
}
|
||
return 0, err
|
||
}
|
||
rv, _ := strconv.ParseInt(cfg.ResourceVersion, 10, 64)
|
||
rvCount += rv
|
||
}
|
||
|
||
return rvCount, nil
|
||
}
|