YuzuZensai/Minikura
1
0
Fork 0
mirror of https://github.com/YuzuZensai/Minikura.git synced 2026-09-13 10:49:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 fix: harden operator reconciliation and install

Browse Source
This commit is contained in:
yuzu
2026-08-13 03:29:29 +07:00
parent ab662a4fa4
commit e90dc8382b
28 changed files with 800 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
operator/internal/controller/apply.go
+31
View File
@@ -3,6 +3,7 @@ package controller
import (
"context"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
@@ -16,6 +17,11 @@ var applyOpts = []client.PatchOption{
}
func apply(ctx context.Context, c client.Client, owner client.Object, obj client.Object, scheme *runtime.Scheme) error {
if svc, ok := obj.(*corev1.Service); ok {
if err := preserveServiceAllocations(ctx, c, svc); err != nil {
return err
}
}
if err := setOwner(owner, obj, scheme); err != nil {
return err
}
@@ -30,3 +36,28 @@ func apply(ctx context.Context, c client.Client, owner client.Object, obj client
obj.SetResourceVersion("")
return c.Patch(ctx, obj, client.Apply, applyOpts...)
}
func preserveServiceAllocations(ctx context.Context, c client.Client, desired *corev1.Service) error {
var current corev1.Service
if err := c.Get(ctx, client.ObjectKeyFromObject(desired), &current); err != nil {
return client.IgnoreNotFound(err)
}
desired.Spec.ClusterIP = current.Spec.ClusterIP
desired.Spec.ClusterIPs = append([]string(nil), current.Spec.ClusterIPs...)
desired.Spec.IPFamilies = append([]corev1.IPFamily(nil), current.Spec.IPFamilies...)
desired.Spec.IPFamilyPolicy = current.Spec.IPFamilyPolicy
if desired.Spec.Type != corev1.ServiceTypeClusterIP {
for i := range desired.Spec.Ports {
if desired.Spec.Ports[i].NodePort != 0 {
continue
}
for _, port := range current.Spec.Ports {
if port.Name == desired.Spec.Ports[i].Name && port.Protocol == desired.Spec.Ports[i].Protocol {
desired.Spec.Ports[i].NodePort = port.NodePort
break
}
}
}
}
return nil
}
operator/internal/controller/apply_test.go
+22
View File
@@ -11,6 +11,7 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
v1alpha1 "github.com/YuzuZensai/Minikura/operator/api/v1alpha1"
"github.com/YuzuZensai/Minikura/operator/internal/resources"
)
func TestTypedObjectMarshalsWithoutTypeMeta(t *testing.T) {
@@ -33,6 +34,27 @@ func TestTypedObjectMarshalsWithoutTypeMeta(t *testing.T) {
}
}
func TestApplyPreservesServiceAllocations(t *testing.T) {
owner := testMinecraft("smp", v1alpha1.ServerStateless)
current := resources.MinecraftService(owner)
current.Spec.ClusterIP = "10.0.0.10"
current.Spec.ClusterIPs = []string{"10.0.0.10"}
current.Spec.IPFamilies = []corev1.IPFamily{corev1.IPv4Protocol}
policy := corev1.IPFamilyPolicySingleStack
current.Spec.IPFamilyPolicy = &policy
c := newFakeClient(t, owner, current)
desired := resources.MinecraftService(owner)
if err := apply(context.Background(), c, owner, desired, testScheme(t)); err != nil {
t.Fatalf("apply: %v", err)
}
var got corev1.Service
mustGet(t, c, client.ObjectKeyFromObject(current), &got)
if got.Spec.ClusterIP != "10.0.0.10" || len(got.Spec.ClusterIPs) != 1 {
t.Errorf("service allocations were not preserved: %+v", got.Spec)
}
}
func TestApplySetsTypeMeta(t *testing.T) {
scheme := testScheme(t)
owner := &v1alpha1.MinecraftServer{
operator/internal/controller/minecraftserver_controller.go
+13 -5
View File
@@ -2,6 +2,7 @@ package controller
import (
"context"
"fmt"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
@@ -10,7 +11,6 @@ import (
"k8s.io/apimachinery/pkg/runtime"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
v1alpha1 "github.com/YuzuZensai/Minikura/operator/api/v1alpha1"
"github.com/YuzuZensai/Minikura/operator/internal/resources"
@@ -31,8 +31,6 @@ type MinecraftServerReconciler struct {
// +kubebuilder:rbac:groups="",resources=events,verbs=create;patch
func (r *MinecraftServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
logger := log.FromContext(ctx)
var mc v1alpha1.MinecraftServer
if err := r.Get(ctx, req.NamespacedName, &mc); err != nil {
return ctrl.Result{}, client.IgnoreNotFound(err)
@@ -41,6 +39,12 @@ func (r *MinecraftServerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
if !mc.DeletionTimestamp.IsZero() {
return ctrl.Result{}, nil
}
if mc.Spec.Type != v1alpha1.ServerStateful && mc.Spec.Type != v1alpha1.ServerStateless {
return r.fail(ctx, &mc, "InvalidSpec", fmt.Errorf("unsupported server type %q", mc.Spec.Type))
}
if mc.Spec.ListenPort <= 0 {
return r.fail(ctx, &mc, "InvalidSpec", fmt.Errorf("listenPort must be set"))
}
if err := apply(ctx, r.Client, &mc, resources.MinecraftConfigMap(&mc), r.Scheme); err != nil {
return r.fail(ctx, &mc, "ConfigMapFailed", err)
@@ -56,10 +60,13 @@ func (r *MinecraftServerReconciler) Reconcile(ctx context.Context, req ctrl.Requ
}
if err := r.pruneOppositeWorkload(ctx, &mc, stateful); err != nil {
logger.Error(err, "failed to prune previous workload")
return r.fail(ctx, &mc, "PruneFailed", err)
}
return ctrl.Result{}, r.updateStatus(ctx, &mc, stateful)
if err := r.updateStatus(ctx, &mc, stateful); err != nil {
return r.fail(ctx, &mc, "StatusFailed", err)
}
return ctrl.Result{}, nil
}
func (r *MinecraftServerReconciler) reconcileWorkload(ctx context.Context, mc *v1alpha1.MinecraftServer, stateful bool) error {
@@ -155,6 +162,7 @@ func (r *MinecraftServerReconciler) fail(ctx context.Context, mc *v1alpha1.Minec
return failWithStatus(ctx, r.Client, mc, cause, func() {
mc.Status.Phase = v1alpha1.PhaseFailed
mc.Status.Message = cause.Error()
mc.Status.ObservedGeneration = mc.Generation
setCondition(&mc.Status.Conditions, metav1.Condition{
Type: v1alpha1.ConditionReady,
Status: metav1.ConditionFalse,
operator/internal/controller/minecraftserver_controller_test.go
+57 -3
View File
@@ -54,6 +54,24 @@ func TestMinecraftReconcileStatelessCreatesResources(t *testing.T) {
}
}
func TestMinecraftReconcileRejectsInvalidSpec(t *testing.T) {
mc := testMinecraft("invalid", v1alpha1.ServerStateless)
mc.Spec.Type = ""
mc.Spec.ListenPort = 0
c := newFakeClient(t, mc)
r := &MinecraftServerReconciler{Client: c, Scheme: testScheme(t)}
if _, err := r.Reconcile(context.Background(), requestFor(mc)); err == nil || !strings.Contains(err.Error(), "unsupported server type") {
t.Fatalf("reconcile error = %v, want invalid spec error", err)
}
var got v1alpha1.MinecraftServer
mustGet(t, c, client.ObjectKeyFromObject(mc), &got)
if got.Status.Phase != v1alpha1.PhaseFailed || got.Status.ObservedGeneration != mc.Generation {
t.Fatalf("status = %#v, want failed observed generation", got.Status)
}
}
func TestMinecraftReconcileStatefulCreatesStatefulSet(t *testing.T) {
mc := testMinecraft("smp", v1alpha1.ServerStateful)
c := newFakeClient(t, mc)
@@ -156,6 +174,9 @@ func TestMinecraftReconcileInvalidStorageFails(t *testing.T) {
if got.Status.Message == "" {
t.Error("expected a failure message")
}
if got.Status.ObservedGeneration != mc.Generation {
t.Errorf("observedGeneration = %d, want %d", got.Status.ObservedGeneration, mc.Generation)
}
}
func TestMinecraftPruneSkipsUnownedWorkload(t *testing.T) {
@@ -212,6 +233,39 @@ func TestMinecraftStatusUsesReadyReplicas(t *testing.T) {
}
}
func TestMinecraftReconcileMutatesServiceAndWorkload(t *testing.T) {
mc := testMinecraft("mutable", v1alpha1.ServerStateless)
c := newFakeClient(t, mc)
r := &MinecraftServerReconciler{Client: c, Scheme: testScheme(t)}
if _, err := r.Reconcile(context.Background(), requestFor(mc)); err != nil {
t.Fatalf("first reconcile: %v", err)
}
var current v1alpha1.MinecraftServer
mustGet(t, c, client.ObjectKeyFromObject(mc), &current)
current.Spec.ListenPort = 25566
current.Spec.Resources.MemoryLimitMB = 4096
current.Generation = 2
if err := c.Update(context.Background(), &current); err != nil {
t.Fatalf("update server: %v", err)
}
if _, err := r.Reconcile(context.Background(), requestFor(&current)); err != nil {
t.Fatalf("second reconcile: %v", err)
}
key := client.ObjectKey{Name: resources.ServerName(mc.Name), Namespace: mc.Namespace}
var svc corev1.Service
mustGet(t, c, key, &svc)
if svc.Spec.Ports[0].Port != 25566 {
t.Errorf("service port = %d", svc.Spec.Ports[0].Port)
}
var dep appsv1.Deployment
mustGet(t, c, key, &dep)
if got := dep.Spec.Template.Spec.Containers[0].Resources.Limits.Memory().String(); got != "4Gi" {
t.Errorf("memory limit = %q", got)
}
}
func TestMinecraftReconcileApplyFailure(t *testing.T) {
mc := testMinecraft("lobby", v1alpha1.ServerStateless)
c := newInterceptedClient(t, interceptor.Funcs{
@@ -254,7 +308,7 @@ func TestMinecraftReconcileServiceFailure(t *testing.T) {
}
}
func TestMinecraftReconcileContinuesAfterPruneError(t *testing.T) {
func TestMinecraftReconcileReportsPruneError(t *testing.T) {
mc := testMinecraft("lobby", v1alpha1.ServerStateless)
c := newInterceptedClient(t, interceptor.Funcs{
Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
@@ -265,8 +319,8 @@ func TestMinecraftReconcileContinuesAfterPruneError(t *testing.T) {
},
}, mc)
r := &MinecraftServerReconciler{Client: c, Scheme: testScheme(t)}
if _, err := r.Reconcile(context.Background(), requestFor(mc)); err != nil {
t.Fatalf("prune errors should not fail reconcile: %v", err)
if _, err := r.Reconcile(context.Background(), requestFor(mc)); err == nil {
t.Fatal("expected prune failure")
}
}
operator/internal/controller/reverseproxyserver_controller.go
+11 -1
View File
@@ -41,6 +41,12 @@ func (r *ReverseProxyServerReconciler) Reconcile(ctx context.Context, req ctrl.R
if !rp.DeletionTimestamp.IsZero() {
return ctrl.Result{}, nil
}
if rp.Spec.Type != v1alpha1.ProxyVelocity && rp.Spec.Type != v1alpha1.ProxyBungeeCord {
return r.fail(ctx, &rp, "InvalidSpec", fmt.Errorf("unsupported proxy type %q", rp.Spec.Type))
}
if rp.Spec.ExternalPort <= 0 || rp.Spec.ListenPort <= 0 {
return r.fail(ctx, &rp, "InvalidSpec", fmt.Errorf("externalPort and listenPort must be set"))
}
if err := apply(ctx, r.Client, &rp, resources.ProxyConfigMap(&rp), r.Scheme); err != nil {
return r.fail(ctx, &rp, "ConfigMapFailed", err)
@@ -57,7 +63,10 @@ func (r *ReverseProxyServerReconciler) Reconcile(ctx context.Context, req ctrl.R
return r.fail(ctx, &rp, "PruneFailed", err)
}
return ctrl.Result{}, r.updateStatus(ctx, &rp)
if err := r.updateStatus(ctx, &rp); err != nil {
return r.fail(ctx, &rp, "StatusFailed", err)
}
return ctrl.Result{}, nil
}
func (r *ReverseProxyServerReconciler) pruneStaleResources(ctx context.Context, rp *v1alpha1.ReverseProxyServer) error {
@@ -160,6 +169,7 @@ func (r *ReverseProxyServerReconciler) fail(ctx context.Context, rp *v1alpha1.Re
return failWithStatus(ctx, r.Client, rp, cause, func() {
rp.Status.Phase = v1alpha1.PhaseFailed
rp.Status.Message = cause.Error()
rp.Status.ObservedGeneration = rp.Generation
setCondition(&rp.Status.Conditions, metav1.Condition{
Type: v1alpha1.ConditionReady,
Status: metav1.ConditionFalse,
operator/internal/controller/reverseproxyserver_controller_test.go
+59
View File
@@ -45,6 +45,25 @@ func TestProxyReconcileCreatesResources(t *testing.T) {
}
}
func TestProxyReconcileRejectsInvalidSpec(t *testing.T) {
rp := testProxy("invalid", v1alpha1.ProxyVelocity)
rp.Spec.Type = ""
rp.Spec.ExternalPort = 0
rp.Spec.ListenPort = 0
c := newFakeClient(t, rp)
r := &ReverseProxyServerReconciler{Client: c, Scheme: testScheme(t)}
if _, err := r.Reconcile(context.Background(), requestFor(rp)); err == nil || !strings.Contains(err.Error(), "unsupported proxy type") {
t.Fatalf("reconcile error = %v, want invalid spec error", err)
}
var got v1alpha1.ReverseProxyServer
mustGet(t, c, client.ObjectKeyFromObject(rp), &got)
if got.Status.Phase != v1alpha1.PhaseFailed || got.Status.ObservedGeneration != rp.Generation {
t.Fatalf("status = %#v, want failed observed generation", got.Status)
}
}
func TestProxyReconcilePrunesStaleType(t *testing.T) {
rp := testProxy("edge", v1alpha1.ProxyVelocity)
c := newFakeClient(t, rp)
@@ -447,6 +466,46 @@ func TestProxyStatusReadyAndEndpoint(t *testing.T) {
}
}
func TestProxyReconcileMutatesServiceAndDeployment(t *testing.T) {
rp := testProxy("mutable", v1alpha1.ProxyVelocity)
c := newFakeClient(t, rp)
r := &ReverseProxyServerReconciler{Client: c, Scheme: testScheme(t)}
if _, err := r.Reconcile(context.Background(), requestFor(rp)); err != nil {
t.Fatalf("first reconcile: %v", err)
}
var current v1alpha1.ReverseProxyServer
mustGet(t, c, client.ObjectKeyFromObject(rp), &current)
current.Spec.ExternalPort = 25570
current.Spec.BackendURL = "http://backend:3000/api"
current.Spec.APIKeySecretRef = "proxy-key"
current.Generation = 2
if err := c.Update(context.Background(), &current); err != nil {
t.Fatalf("update proxy: %v", err)
}
if _, err := r.Reconcile(context.Background(), requestFor(&current)); err != nil {
t.Fatalf("second reconcile: %v", err)
}
key := client.ObjectKey{Name: resources.ProxyName(rp.Spec.Type, rp.Name), Namespace: rp.Namespace}
var svc corev1.Service
mustGet(t, c, key, &svc)
if svc.Spec.Ports[0].Port != 25570 {
t.Errorf("service port = %d", svc.Spec.Ports[0].Port)
}
var dep appsv1.Deployment
mustGet(t, c, key, &dep)
found := false
for _, env := range dep.Spec.Template.Spec.Containers[0].Env {
if env.Name == "MINIKURA_API_URL" && env.Value == "http://backend:3000/api" {
found = true
}
}
if !found {
t.Error("deployment was not updated with backend URL")
}
}
func TestProxiesForServerListError(t *testing.T) {
c := newInterceptedClient(t, interceptor.Funcs{
List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {