Implement scaling latency metrics through revisions

neonvm/apis/neonvm/v1/virtualmachine_types.go

@@ -21,8 +21,10 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"time"
 
 	"github.com/samber/lo"
+	"go.uber.org/zap/zapcore"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -139,6 +141,15 @@ type VirtualMachineSpec struct {
 	// +kubebuilder:default:=true
 	// +optional
 	EnableSSH *bool `json:"enableSSH,omitempty"`
+
+	// TargetRevision is the identifier set by external party to track when changes to the spec
+	// propagate to the VM.
+	//
+	// If a certain value is written into Spec.TargetRevision together with the changes, and
+	// the same value is observed in Status.CurrentRevision, it means that the changes were
+	// propagated to the VM.
+	// +optional
+	TargetRevision *RevisionWithTime `json:"targetRevision,omitempty"`
 }
 
 func (spec *VirtualMachineSpec) Resources() VirtualMachineResources {
@@ -215,6 +226,66 @@ func (g Guest) ValidateForMemoryProvider(p MemoryProvider) error {
 	return nil
 }
 
+// Flag is a bitmask of flags. The meaning is up to the user.
+//
+// Used in Revision below.
+type Flag uint64
+
+func (f *Flag) Set(flag Flag) {
+	*f |= flag
+}
+
+func (f *Flag) Clear(flag Flag) {
+	*f &= ^flag
+}
+
+func (f *Flag) Has(flag Flag) bool {
+	return *f&flag != 0
+}
+
+// Revision is an identifier, which can be assigned to a specific configuration of a VM.
+// Later it can be used to track the application of the configuration.
+type Revision struct {
+	Value int64 `json:"value"`
+	Flags Flag  `json:"flags"`
+}
+
+// ZeroRevision is the default value when revisions updates are disabled.
+var ZeroRevision = Revision{Value: 0, Flags: 0}
+
+func (r Revision) Min(other Revision) Revision {
+	if r.Value < other.Value {
+		return r
+	}
+	return other
+}
+
+func (r Revision) WithTime(t time.Time) RevisionWithTime {
+	return RevisionWithTime{
+		Revision:  r,
+		UpdatedAt: metav1.NewTime(t),
+	}
+}
+
+// MarshalLogObject implements zapcore.ObjectMarshaler, so that Revision can be used with zap.Object
+func (r *Revision) MarshalLogObject(enc zapcore.ObjectEncoder) error {
+	enc.AddInt64("value", r.Value)
+	enc.AddUint64("flags", uint64(r.Flags))
+	return nil
+}
+
+// RevisionWithTime contains a Revision and the time it was last updated.
+type RevisionWithTime struct {
+	Revision  `json:"revision"`
+	UpdatedAt metav1.Time `json:"updatedAt"`
+}
+
+// MarshalLogObject implements zapcore.ObjectMarshaler, so that RevisionWithTime can be used with zap.Object
+func (r *RevisionWithTime) MarshalLogObject(enc zapcore.ObjectEncoder) error {
+	enc.AddTime("updatedAt", r.UpdatedAt.Time)
+	return r.Revision.MarshalLogObject(enc)
+}
+
 type GuestSettings struct {
 	// Individual lines to add to a sysctl.conf file. See sysctl.conf(5) for more
 	// +optional
@@ -534,6 +605,11 @@ type VirtualMachineStatus struct {
 	MemoryProvider *MemoryProvider `json:"memoryProvider,omitempty"`
 	// +optional
 	SSHSecretName string `json:"sshSecretName,omitempty"`
+
+	// CurrentRevision is updated with Spec.TargetRevision's value once
+	// the changes are propagated to the VM.
+	// +optional
+	CurrentRevision *RevisionWithTime `json:"currentRevision,omitempty"`
 }
 
 type VmPhase string

neonvm/config/crd/bases/vm.neon.tech_virtualmachines.yaml

@@ -2662,6 +2662,37 @@ spec:
                 type: boolean
               serviceAccountName:
                 type: string
+              targetRevision:
+                description: "TargetRevision is the identifier set by external party
+                  to track when changes to the spec propagate to the VM. \n If a certain
+                  value is written into Spec.TargetRevision together with the changes,
+                  and the same value is observed in Status.CurrentRevision, it means
+                  that the changes were propagated to the VM."
+                properties:
+                  revision:
+                    description: Revision is an identifier, which can be assigned
+                      to a specific configuration of a VM. Later it can be used to
+                      track the application of the configuration.
+                    properties:
+                      flags:
+                        description: "Flag is a bitmask of flags. The meaning is up
+                          to the user. \n Used in Revision below."
+                        format: int64
+                        type: integer
+                      value:
+                        format: int64
+                        type: integer
+                    required:
+                    - flags
+                    - value
+                    type: object
+                  updatedAt:
+                    format: date-time
+                    type: string
+                required:
+                - revision
+                - updatedAt
+                type: object
               terminationGracePeriodSeconds:
                 default: 5
                 format: int64
@@ -2786,6 +2817,34 @@ spec:
                 pattern: ^[0-9]+((\.[0-9]*)?|m)
                 type: integer
                 x-kubernetes-int-or-string: true
+              currentRevision:
+                description: CurrentRevision is updated with Spec.TargetRevision's
+                  value once the changes are propagated to the VM.
+                properties:
+                  revision:
+                    description: Revision is an identifier, which can be assigned
+                      to a specific configuration of a VM. Later it can be used to
+                      track the application of the configuration.
+                    properties:
+                      flags:
+                        description: "Flag is a bitmask of flags. The meaning is up
+                          to the user. \n Used in Revision below."
+                        format: int64
+                        type: integer
+                      value:
+                        format: int64
+                        type: integer
+                    required:
+                    - flags
+                    - value
+                    type: object
+                  updatedAt:
+                    format: date-time
+                    type: string
+                required:
+                - revision
+                - updatedAt
+                type: object
               extraNetIP:
                 type: string
               extraNetMask:

neonvm/controllers/vm_controller.go

@@ -807,9 +807,27 @@ func (r *VMReconciler) doReconcile(ctx context.Context, vm *vmv1.VirtualMachine)
 		// do nothing
 	}
 
+	// Propagate TargetRevision to CurrentRevision. This is done only if the VM is fully
+	// reconciled and running.
+	if vm.Status.Phase == vmv1.VmRunning {
+		propagateRevision(vm)
+	}
+
 	return nil
 }
 
+func propagateRevision(vm *vmv1.VirtualMachine) {
+	if vm.Spec.TargetRevision == nil {
+		return
+	}
+	if vm.Status.CurrentRevision != nil &&
+		vm.Status.CurrentRevision.Revision == vm.Spec.TargetRevision.Revision {
+		return
+	}
+	rev := vm.Spec.TargetRevision.WithTime(time.Now())
+	vm.Status.CurrentRevision = &rev
+}
+
 func pickMemoryProvider(config *ReconcilerConfig, vm *vmv1.VirtualMachine) vmv1.MemoryProvider {
 	if p := vm.Spec.Guest.MemoryProvider; p != nil {
 		return *p

pkg/agent/core/action.go

@@ -5,6 +5,7 @@ import (
 
 	"go.uber.org/zap/zapcore"
 
+	vmv1 "github.com/neondatabase/autoscaling/neonvm/apis/neonvm/v1"
 	"github.com/neondatabase/autoscaling/pkg/api"
 )
 
@@ -21,24 +22,28 @@ type ActionWait struct {
 }
 
 type ActionPluginRequest struct {
-	LastPermit *api.Resources `json:"current"`
-	Target     api.Resources  `json:"target"`
-	Metrics    *api.Metrics   `json:"metrics"`
+	LastPermit     *api.Resources        `json:"current"`
+	Target         api.Resources         `json:"target"`
+	Metrics        *api.Metrics          `json:"metrics"`
+	TargetRevision vmv1.RevisionWithTime `json:"targetRevision"`
 }
 
 type ActionNeonVMRequest struct {
-	Current api.Resources `json:"current"`
-	Target  api.Resources `json:"target"`
+	Current        api.Resources         `json:"current"`
+	Target         api.Resources         `json:"target"`
+	TargetRevision vmv1.RevisionWithTime `json:"targetRevision"`
 }
 
 type ActionMonitorDownscale struct {
-	Current api.Resources `json:"current"`
-	Target  api.Resources `json:"target"`
+	Current        api.Resources         `json:"current"`
+	Target         api.Resources         `json:"target"`
+	TargetRevision vmv1.RevisionWithTime `json:"targetRevision"`
 }
 
 type ActionMonitorUpscale struct {
-	Current api.Resources `json:"current"`
-	Target  api.Resources `json:"target"`
+	Current        api.Resources         `json:"current"`
+	Target         api.Resources         `json:"target"`
+	TargetRevision vmv1.RevisionWithTime `json:"targetRevision"`
 }
 
 func addObjectPtr[T zapcore.ObjectMarshaler](enc zapcore.ObjectEncoder, key string, value *T) error {

pkg/agent/core/dumpstate.go

@@ -33,23 +33,26 @@ func (d StateDump) MarshalJSON() ([]byte, error) {
 func (s *State) Dump() StateDump {
 	return StateDump{
 		internal: state{
-			Debug:   s.internal.Debug,
-			Config:  s.internal.Config,
-			VM:      s.internal.VM,
-			Plugin:  s.internal.Plugin.deepCopy(),
-			Monitor: s.internal.Monitor.deepCopy(),
-			NeonVM:  s.internal.NeonVM.deepCopy(),
-			Metrics: shallowCopy[SystemMetrics](s.internal.Metrics),
+			Debug:                s.internal.Debug,
+			Config:               s.internal.Config,
+			VM:                   s.internal.VM,
+			Plugin:               s.internal.Plugin.deepCopy(),
+			Monitor:              s.internal.Monitor.deepCopy(),
+			NeonVM:               s.internal.NeonVM.deepCopy(),
+			Metrics:              shallowCopy[SystemMetrics](s.internal.Metrics),
+			TargetRevision:       s.internal.TargetRevision,
+			LastDesiredResources: s.internal.LastDesiredResources,
 		},
 	}
 }
 
 func (s *pluginState) deepCopy() pluginState {
 	return pluginState{
-		OngoingRequest: s.OngoingRequest,
-		LastRequest:    shallowCopy[pluginRequested](s.LastRequest),
-		LastFailureAt:  shallowCopy[time.Time](s.LastFailureAt),
-		Permit:         shallowCopy[api.Resources](s.Permit),
+		OngoingRequest:  s.OngoingRequest,
+		LastRequest:     shallowCopy[pluginRequested](s.LastRequest),
+		LastFailureAt:   shallowCopy[time.Time](s.LastFailureAt),
+		Permit:          shallowCopy[api.Resources](s.Permit),
+		CurrentRevision: s.CurrentRevision,
 	}
 }
 
@@ -61,6 +64,7 @@ func (s *monitorState) deepCopy() monitorState {
 		Approved:           shallowCopy[api.Resources](s.Approved),
 		DownscaleFailureAt: shallowCopy[time.Time](s.DownscaleFailureAt),
 		UpscaleFailureAt:   shallowCopy[time.Time](s.UpscaleFailureAt),
+		CurrentRevision:    s.CurrentRevision,
 	}
 }
 
@@ -69,5 +73,7 @@ func (s *neonvmState) deepCopy() neonvmState {
 		LastSuccess:      shallowCopy[api.Resources](s.LastSuccess),
 		OngoingRequested: shallowCopy[api.Resources](s.OngoingRequested),
 		RequestFailedAt:  shallowCopy[time.Time](s.RequestFailedAt),
+		TargetRevision:   s.TargetRevision,
+		CurrentRevision:  s.CurrentRevision,
 	}
 }