[mirror] add coordination lease heartbeat and mirror protection types

Add coordination.k8s.io/v1 Lease resource type for the Apoxy apiserver
and implement heartbeat lease renewal in the mirror controller. The
kube-controller now renews a coordination Lease every 10s (30s duration)
so the cloud-side GC controller can detect disconnected mirrors and
clean up orphaned objects. All sync methods now stamp a heartbeat
annotation as a secondary liveness signal.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dilyevsky

api/coordination/v1/doc.go

@@ -0,0 +1,5 @@
+// +k8s:deepcopy-gen=package
+// +k8s:openapi-gen=true
+
+// Package v1 contains the coordination.k8s.io/v1 Lease API type for the Apoxy apiserver.
+package v1

api/coordination/v1/register.go

@@ -0,0 +1,43 @@
+package v1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// GroupName specifies the group name used to register the objects.
+const GroupName = "coordination.k8s.io"
+
+// GroupVersion specifies the group and the version used to register the objects.
+var GroupVersion = metav1.GroupVersion{Group: GroupName, Version: "v1"}
+
+// SchemeGroupVersion is group version used to register these objects.
+var SchemeGroupVersion = schema.GroupVersion{Group: GroupName, Version: "v1"}
+
+// Resource takes an unqualified resource and returns a Group qualified GroupResource.
+func Resource(resource string) schema.GroupResource {
+	return SchemeGroupVersion.WithResource(resource).GroupResource()
+}
+
+var (
+	// SchemeBuilder and AddToScheme will stay in k8s.io/kubernetes.
+	SchemeBuilder      runtime.SchemeBuilder
+	localSchemeBuilder = &SchemeBuilder
+	// AddToScheme adds the types in this group-version to the given scheme.
+	AddToScheme = localSchemeBuilder.AddToScheme
+)
+
+func init() {
+	localSchemeBuilder.Register(addKnownTypes)
+}
+
+// Adds the list of known types to Scheme.
+func addKnownTypes(scheme *runtime.Scheme) error {
+	scheme.AddKnownTypes(SchemeGroupVersion,
+		&Lease{},
+		&LeaseList{},
+	)
+	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
+	return nil
+}

api/coordination/v1/types.go

@@ -0,0 +1,74 @@
+package v1
+
+import (
+	coordinationv1 "k8s.io/api/coordination/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/apiserver-runtime/pkg/builder/resource"
+)
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +k8s:openapi-gen=true
+// +kubebuilder:object:root=true
+
+// Lease is a wrapper around the standard coordination.k8s.io/v1 Lease that
+// implements the apiserver-runtime resource.Object interface. It reuses the
+// upstream LeaseSpec so clients can use the standard coordination API.
+type Lease struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec coordinationv1.LeaseSpec `json:"spec,omitempty"`
+}
+
+// Implement resource.Object interface for apiserver-runtime.
+var _ resource.Object = &Lease{}
+
+func (l *Lease) GetObjectMeta() *metav1.ObjectMeta {
+	return &l.ObjectMeta
+}
+
+func (l *Lease) NamespaceScoped() bool {
+	return true
+}
+
+func (l *Lease) New() runtime.Object {
+	return &Lease{}
+}
+
+func (l *Lease) NewList() runtime.Object {
+	return &LeaseList{}
+}
+
+func (l *Lease) GetGroupVersionResource() schema.GroupVersionResource {
+	return schema.GroupVersionResource{
+		Group:    SchemeGroupVersion.Group,
+		Version:  SchemeGroupVersion.Version,
+		Resource: "leases",
+	}
+}
+
+func (l *Lease) IsStorageVersion() bool {
+	return true
+}
+
+func (l *Lease) GetSingularName() string {
+	return "lease"
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +kubebuilder:object:root=true
+
+// LeaseList contains a list of Lease objects.
+type LeaseList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []Lease `json:"items"`
+}
+
+var _ resource.ObjectList = &LeaseList{}
+
+func (ll *LeaseList) GetListMeta() *metav1.ListMeta {
+	return &ll.ListMeta
+}

api/coordination/v1/zz_generated.deepcopy.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"fmt"
 
+	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	coordinationclient "k8s.io/client-go/kubernetes/typed/coordination/v1"
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
@@ -42,10 +44,18 @@ func runKubeMirror(ctx context.Context, cfg *configv1alpha1.Config, mc *configv1
 		return fmt.Errorf("failed to create Apoxy client: %w", err)
 	}
 
-	reconciler := controllers.NewMirrorReconciler(mgr.GetClient(), apoxyClient, mc)
+	coordClient, err := coordinationclient.NewForConfig(kCluster)
+	if err != nil {
+		return fmt.Errorf("failed to create coordination client: %w", err)
+	}
+
+	reconciler := controllers.NewMirrorReconciler(mgr.GetClient(), apoxyClient, coordClient, mc)
 	if err := reconciler.SetupWithManager(ctx, mgr); err != nil {
 		return fmt.Errorf("failed to setup mirror reconciler: %w", err)
 	}
 
-	return mgr.Start(ctx)
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error { return mgr.Start(ctx) })
+	g.Go(func() error { return reconciler.RunHeartbeat(ctx, mc.Namespace) })
+	return g.Wait()
 }