diff --git a/controllers/functionconfigs/reconciler/functionconfigreconciler.go b/controllers/functionconfigs/functionconfigreconciler.go similarity index 94% rename from controllers/functionconfigs/reconciler/functionconfigreconciler.go rename to controllers/functionconfigs/functionconfigreconciler.go index 67e2e6ce1..8872a2a35 100644 --- a/controllers/functionconfigs/reconciler/functionconfigreconciler.go +++ b/controllers/functionconfigs/functionconfigreconciler.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package reconciler +package functionconfigs import ( "context" @@ -35,6 +35,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/predicate" ) const BaseFinalizer = "config.porch.kpt.dev/functionconfig" @@ -250,7 +251,7 @@ const ( ReconcilerForController ReconcilerFor = "controller" ) -type FunctionConfigReconciler struct { +type Reconciler struct { Client client.Client FunctionConfigStore *FunctionConfigStore // For indicates which component the reconciler is collecting the configs for @@ -258,7 +259,14 @@ type FunctionConfigReconciler struct { For ReconcilerFor } -func (r *FunctionConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, finalErr error) { +func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + For(&configapi.FunctionConfig{}). + WithEventFilter(predicate.GenerationChangedPredicate{}). + Complete(r) +} + +func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, finalErr error) { klog.Infof("FunctionConfig %q changed", req.NamespacedName) obj := &configapi.FunctionConfig{} err := r.Client.Get(ctx, req.NamespacedName, obj) @@ -330,7 +338,7 @@ func (r *FunctionConfigReconciler) Reconcile(ctx context.Context, req ctrl.Reque return ctrl.Result{}, nil } -func (r *FunctionConfigReconciler) removeFinalizer(ctx context.Context, obj *configapi.FunctionConfig) error { +func (r *Reconciler) removeFinalizer(ctx context.Context, obj *configapi.FunctionConfig) error { patch := client.MergeFrom(obj.DeepCopy()) switch r.For { @@ -350,7 +358,7 @@ func (r *FunctionConfigReconciler) removeFinalizer(ctx context.Context, obj *con return nil } -func (r *FunctionConfigReconciler) addFinalizer(ctx context.Context, obj *configapi.FunctionConfig) error { +func (r *Reconciler) addFinalizer(ctx context.Context, obj *configapi.FunctionConfig) error { patch := client.MergeFrom(obj.DeepCopy()) updated := false diff --git a/controllers/functionconfigs/reconciler/functionconfigreconciler_test.go b/controllers/functionconfigs/functionconfigreconciler_test.go similarity index 97% rename from controllers/functionconfigs/reconciler/functionconfigreconciler_test.go rename to controllers/functionconfigs/functionconfigreconciler_test.go index d0daa10ac..13803921f 100644 --- a/controllers/functionconfigs/reconciler/functionconfigreconciler_test.go +++ b/controllers/functionconfigs/functionconfigreconciler_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 The kpt Authors +// Copyright 2026 The kpt Authors // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package reconciler +package functionconfigs import ( "context" @@ -39,7 +39,7 @@ func TestFunctionConfigReconciler(t *testing.T) { type testcase struct { name string objs []client.Object // input: objects to seed the fake client - check func(t *testing.T, reconciler *FunctionConfigReconciler) + check func(t *testing.T, reconciler *Reconciler) requests []string // name set in the reconcile request expectErr bool } @@ -137,7 +137,7 @@ func TestFunctionConfigReconciler(t *testing.T) { name: "FunctionConfig object is stored in FunctionStore after reconciliation", objs: []client.Object{sampleFunctionConfig}, requests: []string{"set-image"}, - check: func(t *testing.T, r *FunctionConfigReconciler) { + check: func(t *testing.T, r *Reconciler) { // Check existence of the functionConfig in cluster got, exists := r.FunctionConfigStore.GetFunctionConfig("set-image") expectedNumberOfFunctions := 1 @@ -152,7 +152,7 @@ func TestFunctionConfigReconciler(t *testing.T) { name: "FunctionConfig object is deleted from FunctionStore after reconciliation", objs: []client.Object{}, requests: []string{"set-image"}, - check: func(t *testing.T, r *FunctionConfigReconciler) { + check: func(t *testing.T, r *Reconciler) { // Check existence of the functionConfig in cluster _, exists := r.FunctionConfigStore.GetFunctionConfig("set-image") assert.False(t, exists, "FunctionConfig 'set-image' should not exist in the store") @@ -162,7 +162,7 @@ func TestFunctionConfigReconciler(t *testing.T) { name: "BinaryExecutorCache is available with image", objs: []client.Object{sampleFunctionConfig}, requests: []string{"set-image"}, - check: func(t *testing.T, r *FunctionConfigReconciler) { + check: func(t *testing.T, r *Reconciler) { expectedKey := "ghcr.io/kptdev/krm-functions-catalog/set-image:v0.1.4" expectedPath := "/functions/set-image" binary, exists := r.FunctionConfigStore.GetBinaryFromCache(expectedKey) @@ -174,7 +174,7 @@ func TestFunctionConfigReconciler(t *testing.T) { name: "BuiltInExecutorCache is available for starlark", objs: []client.Object{builtInSetNamespace, builtInApplyReplacements, builtInStarlarkWithId}, requests: []string{"apply-replacements", "set-namespace", "starlark"}, - check: func(t *testing.T, r *FunctionConfigReconciler) { + check: func(t *testing.T, r *Reconciler) { expectedStarlarkKey := "starlark-id" execFunctions := r.FunctionConfigStore.GetExecCache() @@ -197,7 +197,7 @@ func TestFunctionConfigReconciler(t *testing.T) { c := fake.NewClientBuilder().WithObjects(tt.objs...).WithScheme(scheme).WithStatusSubresource(&configapi.FunctionConfig{}).Build() functionConfigStore := NewFunctionConfigStore(defaultImagePrefix, functionCacheDir) - reconciler := &FunctionConfigReconciler{ + reconciler := &Reconciler{ Client: c, FunctionConfigStore: functionConfigStore, } @@ -260,7 +260,7 @@ func TestFinalizersAdded(t *testing.T) { } c := fake.NewClientBuilder().WithScheme(schemeWithFunctionConfig(t)).WithObjects(obj).WithStatusSubresource(&configapi.FunctionConfig{}).Build() - r := &FunctionConfigReconciler{ + r := &Reconciler{ Client: c, FunctionConfigStore: NewFunctionConfigStore(defaultImagePrefix, functionCacheDir), For: tc.forValue, @@ -528,7 +528,7 @@ func TestFinalizersRemoved(t *testing.T) { store := NewFunctionConfigStore(defaultImagePrefix, functionCacheDir) store.UpsertFunctionConfig(objName, obj) - r := &FunctionConfigReconciler{ + r := &Reconciler{ Client: c, FunctionConfigStore: store, For: tc.forValue, diff --git a/controllers/main.go b/controllers/main.go index fbf9f63eb..2019becec 100644 --- a/controllers/main.go +++ b/controllers/main.go @@ -27,6 +27,7 @@ import ( "slices" + "github.com/kptdev/porch/controllers/functionconfigs" // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) // to ensure that exec-entrypoint and run can make use of them. "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -40,7 +41,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/webhook" "github.com/kptdev/kpt/pkg/lib/runneroptions" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" "github.com/kptdev/porch/controllers/packagerevisions/pkg/controllers/packagerevision" "github.com/kptdev/porch/controllers/packagevariants/pkg/controllers/packagevariant" "github.com/kptdev/porch/controllers/packagevariantsets/pkg/controllers/packagevariantset" @@ -52,7 +52,6 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/healthz" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" porchapi "github.com/kptdev/porch/api/porch/v1alpha1" @@ -305,29 +304,26 @@ func setupReconciler(mgr ctrl.Manager, enabled []string, r Reconciler, started [ return append(started, name), nil } -func setupFunctionConfigReconciler(mgr ctrl.Manager) (*reconciler.FunctionConfigStore, error) { +func setupFunctionConfigReconciler(mgr ctrl.Manager) (*functionconfigs.FunctionConfigStore, error) { prefix := os.Getenv("DEFAULT_IMAGE_PREFIX") if prefix == "" { prefix = runneroptions.GHCRImagePrefix } - functionConfigStore := reconciler.NewFunctionConfigStore(prefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(prefix, "") - rec := &reconciler.FunctionConfigReconciler{ + rec := &functionconfigs.Reconciler{ Client: mgr.GetClient(), FunctionConfigStore: functionConfigStore, - For: reconciler.ReconcilerForController, + For: functionconfigs.ReconcilerForController, } - if err := ctrl.NewControllerManagedBy(mgr). - For(&configapi.FunctionConfig{}). - WithEventFilter(predicate.GenerationChangedPredicate{}). - Complete(rec); err != nil { + if err := rec.SetupWithManager(mgr); err != nil { return nil, fmt.Errorf("error creating FunctionConfig controller: %w", err) } prePopulateFunctionConfigStore(mgr.GetAPIReader(), functionConfigStore) - klog.Infof("FunctionConfig reconciler registered (for: %s)", reconciler.ReconcilerForController) + klog.Infof("FunctionConfig reconciler registered (for: %s)", functionconfigs.ReconcilerForController) return functionConfigStore, nil } @@ -335,7 +331,7 @@ func setupFunctionConfigReconciler(mgr ctrl.Manager) (*reconciler.FunctionConfig // synchronously so the exec cache is ready before the PR controller starts. // Without this, a pod restart leaves the cache empty until the async // informer triggers reconciliation. -func prePopulateFunctionConfigStore(reader client.Reader, store *reconciler.FunctionConfigStore) { +func prePopulateFunctionConfigStore(reader client.Reader, store *functionconfigs.FunctionConfigStore) { var fcList configapi.FunctionConfigList if err := reader.List(context.Background(), &fcList); err != nil { klog.Warningf("FunctionConfig pre-population failed (non-fatal): %v", err) diff --git a/controllers/main_test.go b/controllers/main_test.go index da5005f3d..885a422a8 100644 --- a/controllers/main_test.go +++ b/controllers/main_test.go @@ -20,7 +20,7 @@ import ( "testing" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" mockclient "github.com/kptdev/porch/test/mockery/mocks/external/sigs.k8s.io/controller-runtime/pkg/client" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -171,7 +171,7 @@ func TestPrePopulateFunctionConfigStore_Success(t *testing.T) { list.(*configapi.FunctionConfigList).Items = items }).Return(nil) - store := reconciler.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") + store := functionconfigs.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") prePopulateFunctionConfigStore(mockReader, store) _, ok := store.GetFunctionConfig("set-namespace") @@ -186,7 +186,7 @@ func TestPrePopulateFunctionConfigStore_ListError(t *testing.T) { mockReader := mockclient.NewMockReader(t) mockReader.EXPECT().List(mock.Anything, mock.Anything, mock.Anything).Return(assert.AnError) - store := reconciler.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") + store := functionconfigs.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") prePopulateFunctionConfigStore(mockReader, store) _, ok := store.GetFunctionConfig("anything") @@ -198,7 +198,7 @@ func TestPrePopulateFunctionConfigStore_EmptyList(t *testing.T) { mockReader.EXPECT().List(mock.Anything, mock.AnythingOfType("*v1alpha1.FunctionConfigList"), mock.Anything). Return(nil) - store := reconciler.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") + store := functionconfigs.NewFunctionConfigStore("ghcr.io/kptdev", "/tmp/bins") prePopulateFunctionConfigStore(mockReader, store) assert.Equal(t, 0, len(store.List())) diff --git a/controllers/packagerevisions/pkg/controllers/packagerevision/config_test.go b/controllers/packagerevisions/pkg/controllers/packagerevision/config_test.go index 280cfba90..78d04e5c0 100644 --- a/controllers/packagerevisions/pkg/controllers/packagerevision/config_test.go +++ b/controllers/packagerevisions/pkg/controllers/packagerevision/config_test.go @@ -4,7 +4,7 @@ import ( "flag" "testing" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -62,7 +62,7 @@ func TestInit_NilCache(t *testing.T) { r := &PackageRevisionReconciler{ RepoOperationRetryAttempts: 3, MaxGRPCMessageSize: defaultMaxGRPCMessageSize, - FunctionConfigStore: reconciler.NewFunctionConfigStore("", ""), + FunctionConfigStore: functionconfigs.NewFunctionConfigStore("", ""), } err := r.Init(mgr) require.NoError(t, err) @@ -75,7 +75,7 @@ func TestInit_SetsCredResolverAndFetcher(t *testing.T) { r := &PackageRevisionReconciler{ RepoOperationRetryAttempts: 3, MaxGRPCMessageSize: defaultMaxGRPCMessageSize, - FunctionConfigStore: reconciler.NewFunctionConfigStore("", ""), + FunctionConfigStore: functionconfigs.NewFunctionConfigStore("", ""), } err := r.Init(mgr) @@ -93,7 +93,7 @@ func TestInit_RendererEnabledWithFnRunner(t *testing.T) { r := &PackageRevisionReconciler{ RepoOperationRetryAttempts: 3, MaxGRPCMessageSize: defaultMaxGRPCMessageSize, - FunctionConfigStore: reconciler.NewFunctionConfigStore("", ""), + FunctionConfigStore: functionconfigs.NewFunctionConfigStore("", ""), } t.Setenv("FUNCTION_RUNNER_ADDRESS", "localhost:0") diff --git a/controllers/packagerevisions/pkg/controllers/packagerevision/packagerevision_controller.go b/controllers/packagerevisions/pkg/controllers/packagerevision/packagerevision_controller.go index 68742b782..364a2739d 100644 --- a/controllers/packagerevisions/pkg/controllers/packagerevision/packagerevision_controller.go +++ b/controllers/packagerevisions/pkg/controllers/packagerevision/packagerevision_controller.go @@ -20,7 +20,7 @@ import ( "time" porchv1alpha2 "github.com/kptdev/porch/api/porch/v1alpha2" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" "github.com/kptdev/porch/pkg/repository" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -52,7 +52,7 @@ type PackageRevisionReconciler struct { Scheme *runtime.Scheme ContentCache repository.ContentCache ExternalPackageFetcher repository.ExternalPackageFetcher - FunctionConfigStore *reconciler.FunctionConfigStore + FunctionConfigStore *functionconfigs.FunctionConfigStore Renderer renderer // nil = skip rendering MaxConcurrentReconciles int diff --git a/pkg/apiserver/repo_cache_handler.go b/controllers/repo-cache/repo_cache_handler.go similarity index 84% rename from pkg/apiserver/repo_cache_handler.go rename to controllers/repo-cache/repo_cache_handler.go index eab3ff750..0d4a5761a 100644 --- a/pkg/apiserver/repo_cache_handler.go +++ b/controllers/repo-cache/repo_cache_handler.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package apiserver +package repocache import ( "context" @@ -33,19 +33,21 @@ import ( const repoCacheFinalizer = "config.porch.kpt.dev/porch-server" -// RepoCacheReconciler watches Repository CRs and manages the porch-server +// Reconciler watches Repository CRs and manages the porch-server // in-memory cache: // - On Create/startup: requeues until the repo is Ready, then opens it. // - On spec change (generation bump): re-opens the repository. // - On deletion (DeletionTimestamp set): evicts from cache and removes finalizer. -type RepoCacheReconciler struct { - client client.Client - cache cachetypes.Cache +type Reconciler struct { + Client client.Client + Cache cachetypes.Cache + + MaxConcurrency int } -func (r *RepoCacheReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { +func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { repo := &configapi.Repository{} - if err := r.client.Get(ctx, req.NamespacedName, repo); err != nil { + if err := r.Client.Get(ctx, req.NamespacedName, repo); err != nil { return reconcile.Result{}, client.IgnoreNotFound(err) } @@ -58,14 +60,14 @@ func (r *RepoCacheReconciler) Reconcile(ctx context.Context, req reconcile.Reque if repo.DeletionTimestamp != nil { klog.Infof("Repo cache handler: evicting %s", req.NamespacedName) - if err := r.cache.EvictCachedRepository(ctx, req.Namespace, req.Name); err != nil { + if err := r.Cache.EvictCachedRepository(ctx, req.Namespace, req.Name); err != nil { klog.Warningf("Repo cache handler: failed to evict %s: %v", req.NamespacedName, err) return reconcile.Result{}, err } if controllerutil.ContainsFinalizer(repo, repoCacheFinalizer) { controllerutil.RemoveFinalizer(repo, repoCacheFinalizer) - if err := r.client.Update(ctx, repo); err != nil { + if err := r.Client.Update(ctx, repo); err != nil { return reconcile.Result{}, fmt.Errorf("failed to remove finalizer from %s: %w", req.NamespacedName, err) } } @@ -83,7 +85,7 @@ func (r *RepoCacheReconciler) Reconcile(ctx context.Context, req reconcile.Reque // Add finalizer — guarantees eviction on delete if !controllerutil.ContainsFinalizer(repo, repoCacheFinalizer) { controllerutil.AddFinalizer(repo, repoCacheFinalizer) - if err := r.client.Update(ctx, repo); err != nil { + if err := r.Client.Update(ctx, repo); err != nil { return reconcile.Result{}, fmt.Errorf("failed to add finalizer to %s: %w", req.NamespacedName, err) } klog.Infof("Repo cache handler: added finalizer to %s", req.NamespacedName) @@ -92,7 +94,7 @@ func (r *RepoCacheReconciler) Reconcile(ctx context.Context, req reconcile.Reque // Open the repository in the cache. Internally this uses the cache's // SafeRepoMap.LoadOrCreate, which is idempotent — it returns the existing // map entry if the repo is already cached, so this is a no-op for repeated reconciles. - if _, err := r.cache.OpenRepository(ctx, repo); err != nil { + if _, err := r.Cache.OpenRepository(ctx, repo); err != nil { klog.Warningf("Repo cache handler: failed to open %s: %v", req.NamespacedName, err) return reconcile.Result{}, err } @@ -101,26 +103,36 @@ func (r *RepoCacheReconciler) Reconcile(ctx context.Context, req reconcile.Reque return reconcile.Result{}, nil } -// setupRepoCacheController registers the repo cache controller with the given manager. -func setupRepoCacheController(mgr ctrl.Manager, cache cachetypes.Cache, maxConcurrency int) error { - if maxConcurrency <= 0 { - maxConcurrency = 20 - } - r := &RepoCacheReconciler{ - client: mgr.GetClient(), - cache: cache, +func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error { + if r.Client == nil { + r.Client = mgr.GetClient() } return ctrl.NewControllerManagedBy(mgr). For(&configapi.Repository{}). WithEventFilter(repoCachePredicate()). WithOptions(controller.Options{ - MaxConcurrentReconciles: maxConcurrency, + MaxConcurrentReconciles: r.MaxConcurrency, }). Named("repo-cache"). Complete(r) } +// SetupRepoCacheController registers the repo cache controller with the given manager. +func SetupRepoCacheController(mgr ctrl.Manager, cache cachetypes.Cache, maxConcurrency int) error { + if maxConcurrency <= 0 { + maxConcurrency = 20 + } + r := &Reconciler{ + Client: mgr.GetClient(), + Cache: cache, + + MaxConcurrency: maxConcurrency, + } + + return r.SetupWithManager(mgr) +} + // isRepositoryReady checks if the repo controller has marked this repository as Ready. func isRepositoryReady(repo *configapi.Repository) bool { for _, c := range repo.Status.Conditions { diff --git a/pkg/apiserver/repo_cache_handler_test.go b/controllers/repo-cache/repo_cache_handler_test.go similarity index 86% rename from pkg/apiserver/repo_cache_handler_test.go rename to controllers/repo-cache/repo_cache_handler_test.go index 442d3ae3d..f1b2dc0ac 100644 --- a/pkg/apiserver/repo_cache_handler_test.go +++ b/controllers/repo-cache/repo_cache_handler_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package apiserver +package repocache import ( "context" @@ -28,12 +28,9 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" - "k8s.io/client-go/rest" - ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" "sigs.k8s.io/controller-runtime/pkg/event" - metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) @@ -66,7 +63,7 @@ func TestReconcileReadyRepoOpensAndAddsFinalizer(t *testing.T) { mc := mockcache.NewMockCache(t) mc.EXPECT().OpenRepository(mock.Anything, mock.Anything).Return(nil, nil).Once() - r := &RepoCacheReconciler{client: fakeClient, cache: mc} + r := &Reconciler{Client: fakeClient, Cache: mc} result, err := r.Reconcile(context.Background(), reconcile.Request{ NamespacedName: types.NamespacedName{Name: "my-repo", Namespace: "test-ns"}, @@ -91,7 +88,7 @@ func TestReconcileNotReadyRepoIsSkipped(t *testing.T) { mc := mockcache.NewMockCache(t) // No OpenRepository or EvictCachedRepository calls expected - r := &RepoCacheReconciler{client: fakeClient, cache: mc} + r := &Reconciler{Client: fakeClient, Cache: mc} result, err := r.Reconcile(context.Background(), reconcile.Request{ NamespacedName: types.NamespacedName{Name: "not-ready", Namespace: "test-ns"}, @@ -126,7 +123,7 @@ func TestReconcileDeletingRepoEvictsAndRemovesFinalizer(t *testing.T) { repo: repo, } - r := &RepoCacheReconciler{client: fakeClient, cache: mc} + r := &Reconciler{Client: fakeClient, Cache: mc} result, err := r.Reconcile(context.Background(), reconcile.Request{ NamespacedName: types.NamespacedName{Name: "del-repo", Namespace: "test-ns"}, @@ -165,7 +162,7 @@ func TestReconcileOpenRepositoryError(t *testing.T) { mc := mockcache.NewMockCache(t) mc.EXPECT().OpenRepository(mock.Anything, mock.Anything).Return(nil, fmt.Errorf("connection refused")).Once() - r := &RepoCacheReconciler{client: fakeClient, cache: mc} + r := &Reconciler{Client: fakeClient, Cache: mc} result, err := r.Reconcile(context.Background(), reconcile.Request{ NamespacedName: types.NamespacedName{Name: "fail-repo", Namespace: "test-ns"}, @@ -182,7 +179,7 @@ func TestReconcileNotFoundIsNoOp(t *testing.T) { mc := mockcache.NewMockCache(t) - r := &RepoCacheReconciler{client: fakeClient, cache: mc} + r := &Reconciler{Client: fakeClient, Cache: mc} result, err := r.Reconcile(context.Background(), reconcile.Request{ NamespacedName: types.NamespacedName{Name: "gone", Namespace: "test-ns"}, @@ -201,22 +198,3 @@ func TestRepoCachePredicate(t *testing.T) { assert.False(t, p.Delete(event.DeleteEvent{Object: repo}), "delete should be filtered (handled via finalizer)") assert.False(t, p.Generic(event.GenericEvent{Object: repo}), "generic should be filtered") } - -func TestSetupRepoCacheController(t *testing.T) { - scheme := newTestScheme() - mc := mockcache.NewMockCache(t) - - // Use a fake rest.Config pointing to a non-existent server. - // We only need the manager to accept the controller registration — not actually run. - cfg := &rest.Config{Host: "https://127.0.0.1:0"} - - mgr, err := ctrl.NewManager(cfg, ctrl.Options{ - Scheme: scheme, - Metrics: metricsserver.Options{BindAddress: "0"}, - HealthProbeBindAddress: "0", - }) - require.NoError(t, err) - - err = setupRepoCacheController(mgr, mc, 5) - require.NoError(t, err) -} diff --git a/func/internal/executableevaluator.go b/func/internal/executableevaluator.go index 4e10a4f42..2a16457ed 100644 --- a/func/internal/executableevaluator.go +++ b/func/internal/executableevaluator.go @@ -22,7 +22,7 @@ import ( kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" "github.com/kptdev/kpt/pkg/fn" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" pb "github.com/kptdev/porch/func/evaluator" regclientref "github.com/regclient/regclient/types/ref" "google.golang.org/grpc/codes" @@ -36,12 +36,12 @@ type ExecutableEvaluatorOptions struct { type executableEvaluator struct { // Fast-path function cache - FunctionConfigStore *reconciler.FunctionConfigStore + FunctionConfigStore *functionconfigs.FunctionConfigStore } var _ Evaluator = &executableEvaluator{} -func NewExecutableEvaluator(FunctionConfigStore *reconciler.FunctionConfigStore) (Evaluator, error) { +func NewExecutableEvaluator(FunctionConfigStore *functionconfigs.FunctionConfigStore) (Evaluator, error) { return &executableEvaluator{ FunctionConfigStore: FunctionConfigStore, }, nil diff --git a/func/internal/executableevaluator_test.go b/func/internal/executableevaluator_test.go index 426b27a34..e3b349dae 100644 --- a/func/internal/executableevaluator_test.go +++ b/func/internal/executableevaluator_test.go @@ -25,7 +25,7 @@ import ( kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" "github.com/kptdev/kpt/pkg/fn" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" pb "github.com/kptdev/porch/func/evaluator" imageutil "github.com/kptdev/porch/pkg/util/image" "github.com/stretchr/testify/assert" @@ -39,7 +39,7 @@ const ( starlarkFunction = "starlark" ) -func getFunctionConfigStore(binaryDir string) *reconciler.FunctionConfigStore { +func getFunctionConfigStore(binaryDir string) *functionconfigs.FunctionConfigStore { starlarkConfig := &configapi.FunctionConfig{ Spec: configapi.FunctionConfigSpec{ Image: starlarkFunction, @@ -70,7 +70,7 @@ func getFunctionConfigStore(binaryDir string) *reconciler.FunctionConfigStore { }, }, } - fstore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, binaryDir) + fstore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, binaryDir) fstore.UpdateBinaryCache(starlarkFunction, starlarkConfig) fstore.UpdateBinaryCache(setImageFunction, setImageConfig) return fstore diff --git a/func/internal/podcachemanager.go b/func/internal/podcachemanager.go index 17317f40b..38e7a8a3d 100644 --- a/func/internal/podcachemanager.go +++ b/func/internal/podcachemanager.go @@ -23,7 +23,7 @@ import ( "time" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - fnconf "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + fnconf "github.com/kptdev/porch/controllers/functionconfigs" imageutil "github.com/kptdev/porch/pkg/util/image" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" diff --git a/func/internal/podcachemanager_eventloop_test.go b/func/internal/podcachemanager_eventloop_test.go index 85ca98a7e..d24b7a8af 100644 --- a/func/internal/podcachemanager_eventloop_test.go +++ b/func/internal/podcachemanager_eventloop_test.go @@ -23,7 +23,7 @@ import ( "time" "github.com/kptdev/kpt/pkg/lib/runneroptions" - fnconf "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + fnconf "github.com/kptdev/porch/controllers/functionconfigs" "github.com/stretchr/testify/assert" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" diff --git a/func/internal/podcachemanager_unit_test.go b/func/internal/podcachemanager_unit_test.go index d23a3d95e..806bcd681 100644 --- a/func/internal/podcachemanager_unit_test.go +++ b/func/internal/podcachemanager_unit_test.go @@ -24,7 +24,7 @@ import ( "github.com/kptdev/kpt/pkg/lib/runneroptions" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - fnconf "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + fnconf "github.com/kptdev/porch/controllers/functionconfigs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/grpc" diff --git a/func/internal/podevaluator.go b/func/internal/podevaluator.go index 8cffc2ca6..015b198d3 100644 --- a/func/internal/podevaluator.go +++ b/func/internal/podevaluator.go @@ -21,7 +21,7 @@ import ( "time" "github.com/kptdev/kpt/pkg/fn/runtime" - fnconf "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + fnconf "github.com/kptdev/porch/controllers/functionconfigs" "github.com/kptdev/porch/func/evaluator" "github.com/kptdev/porch/pkg/util" "google.golang.org/grpc" diff --git a/func/internal/podevaluator_podcachemanager_test.go b/func/internal/podevaluator_podcachemanager_test.go index f5defdcec..a6a385928 100644 --- a/func/internal/podevaluator_podcachemanager_test.go +++ b/func/internal/podevaluator_podcachemanager_test.go @@ -27,7 +27,7 @@ import ( "github.com/kptdev/kpt/pkg/lib/runneroptions" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - fnconf "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + fnconf "github.com/kptdev/porch/controllers/functionconfigs" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "k8s.io/apimachinery/pkg/runtime" diff --git a/func/server/server.go b/func/server/server.go index acc776565..df08afb62 100644 --- a/func/server/server.go +++ b/func/server/server.go @@ -26,7 +26,7 @@ import ( "github.com/kptdev/kpt/pkg/lib/runneroptions" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" pb "github.com/kptdev/porch/func/evaluator" "github.com/kptdev/porch/func/healthchecker" "github.com/kptdev/porch/func/internal" @@ -228,7 +228,7 @@ func buildScheme() (*runtime.Scheme, error) { return scheme, nil } -func buildFnConfigReconciler(o *options, scheme *runtime.Scheme) (*reconciler.FunctionConfigReconciler, error) { +func buildFnConfigReconciler(o *options, scheme *runtime.Scheme) (*functionconfigs.Reconciler, error) { restCfg, err := getRestConfig() if err != nil { return nil, err @@ -249,12 +249,12 @@ func buildFnConfigReconciler(o *options, scheme *runtime.Scheme) (*reconciler.Fu return nil, err } - functionConfigStore := reconciler.NewFunctionConfigStore(o.defaultImagePrefix, o.exec.FunctionCacheDir) + functionConfigStore := functionconfigs.NewFunctionConfigStore(o.defaultImagePrefix, o.exec.FunctionCacheDir) - rec := &reconciler.FunctionConfigReconciler{ + rec := &functionconfigs.Reconciler{ Client: mgr.GetClient(), FunctionConfigStore: functionConfigStore, - For: reconciler.ReconcilerForFunctionRunner, + For: functionconfigs.ReconcilerForFunctionRunner, } if err := ctrl.NewControllerManagedBy(mgr). diff --git a/go.mod b/go.mod index 17d811db5..8af4247ef 100644 --- a/go.mod +++ b/go.mod @@ -235,6 +235,7 @@ require ( k8s.io/kube-openapi v0.0.0-20260520065146-aa012df4f4af // indirect k8s.io/streaming v0.36.1 // indirect sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.36.0 // indirect + sigs.k8s.io/cli-utils v0.37.2 // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect sigs.k8s.io/kustomize/api v0.21.1 // indirect sigs.k8s.io/randfill v1.0.0 // indirect diff --git a/pkg/apiserver/apiserver_test.go b/pkg/apiserver/apiserver_test.go index 8f860c672..7782b621c 100644 --- a/pkg/apiserver/apiserver_test.go +++ b/pkg/apiserver/apiserver_test.go @@ -15,22 +15,38 @@ package apiserver import ( + "context" "fmt" + "net" + "os" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" + genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/client-go/rest" + "k8s.io/klog/v2/textlogger" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" ) +const k8sServHostEnv = "KUBERNETES_SERVICE_HOST" +const k8sServPortEnv = "KUBERNETES_SERVICE_PORT" +const loopbackAnyPort = "127.0.0.1:0" +const loopbackPort1 = "127.0.0.1:1" + +func TestMain(m *testing.M) { + ctrllog.SetLogger(textlogger.NewLogger(textlogger.NewConfig())) + os.Exit(m.Run()) +} + func TestBuildCompleteScheme(t *testing.T) { scheme, err := buildCompleteScheme() require.NoError(t, err) require.NotNil(t, scheme) - // Test singleton behavior - calling again should return same instance scheme2, err := buildCompleteScheme() require.NoError(t, err) assert.Same(t, scheme, scheme2, "expected buildCompleteScheme to return singleton instance") @@ -88,3 +104,152 @@ func TestBuildSchemeWithTypes(t *testing.T) { }) } } + +func TestKubeConfig(t *testing.T) { + kubernetesServiceHost := os.Getenv(k8sServHostEnv) + kubernetesServicePort := os.Getenv(k8sServPortEnv) + _ = os.Unsetenv(k8sServHostEnv) + _ = os.Unsetenv(k8sServPortEnv) + + t.Cleanup(func() { + _ = os.Setenv(k8sServHostEnv, kubernetesServiceHost) + _ = os.Setenv(k8sServPortEnv, kubernetesServicePort) + }) + + t.Run("path is empty", func(t *testing.T) { + ln, err := net.Listen("tcp", loopbackAnyPort) + require.NoError(t, err) + defer ln.Close() + + cfg := &Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + } + + completed := cfg.Complete() + assert.NotNil(t, completed.GenericConfig.EffectiveVersion) + assert.Equal(t, "1.0", completed.GenericConfig.EffectiveVersion.BinaryVersion().String()) + + restConfig, err := completed.getRestConfig() + assert.ErrorContains(t, err, "failed to load in-cluster config (specify --kubeconfig if not running in-cluster):") + assert.Nil(t, restConfig) + }) + + t.Run("failed to load config", func(t *testing.T) { + ln, err := net.Listen("tcp", loopbackAnyPort) + require.NoError(t, err) + defer ln.Close() + + cfg := &Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + ExtraConfig: ExtraConfig{ + CoreAPIKubeconfigPath: "/non/existent/path", + }, + } + completed := cfg.Complete() + + restConfig, err := completed.getRestConfig() + assert.ErrorContains(t, err, "failed to load config") + assert.Nil(t, restConfig) + }) + + t.Run("successful buildClient execution", func(t *testing.T) { + ln, err := net.Listen("tcp", loopbackAnyPort) + require.NoError(t, err) + defer ln.Close() + + cfg := &Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + } + completed := cfg.Complete() + + restConfig := &rest.Config{Host: "https://" + loopbackPort1} + scheme, err := buildCompleteScheme() + require.NoError(t, err) + + clientWithWatch, err := completed.buildClient(restConfig, scheme, nil) + require.NoError(t, err) + assert.NotNil(t, clientWithWatch) + }) +} + +func TestGetCoreV1Client(t *testing.T) { + t.Run("getCoreV1Client without any error", func(t *testing.T) { + ln, err := net.Listen("tcp", loopbackAnyPort) + require.NoError(t, err) + defer ln.Close() + + cfg := &Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + } + completed := cfg.Complete() + + restConfig := &rest.Config{Host: "https://" + loopbackPort1} + corev1client, err := completed.getCoreV1Client(restConfig) + assert.NotNil(t, corev1client) + assert.NoError(t, err) + }) +} + +func TestNew(t *testing.T) { + t.Run("uninitialized Serializer", func(t *testing.T) { + ctx := context.Background() + ln, err := net.Listen("tcp", loopbackAnyPort) + require.NoError(t, err) + defer ln.Close() + + cfg := &Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + ExtraConfig: ExtraConfig{ + CoreAPIKubeconfigPath: "/non/existent/path", + }, + } + completed := cfg.Complete() + + _, porchServer, err := completed.New(ctx) + assert.Nil(t, porchServer) + assert.ErrorContains(t, err, "Genericapiserver.New() called with config.Serializer == nil") + }) +} + +func TestZeroToNil(t *testing.T) { + assert.Nil(t, zeroToNil(0)) + assert.Equal(t, new(time.Second), zeroToNil(time.Second)) +} + +func TestPorchServerNeedLeaderElection(t *testing.T) { + s := &PorchServer{leaderElect: true} + assert.True(t, s.NeedLeaderElection()) + + s.leaderElect = false + assert.False(t, s.NeedLeaderElection()) +} diff --git a/pkg/apiserver/apiserver.go b/pkg/apiserver/config.go similarity index 53% rename from pkg/apiserver/apiserver.go rename to pkg/apiserver/config.go index ab765694f..5686a592d 100644 --- a/pkg/apiserver/apiserver.go +++ b/pkg/apiserver/config.go @@ -1,4 +1,4 @@ -// Copyright 2022, 2024-2025 The kpt Authors +// Copyright 2022, 2024-2026 The kpt Authors // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -17,8 +17,6 @@ package apiserver import ( "context" "fmt" - "os" - "strings" "sync" "time" @@ -27,12 +25,12 @@ import ( porchapi "github.com/kptdev/porch/api/porch/v1alpha1" porchv1alpha2 "github.com/kptdev/porch/api/porch/v1alpha2" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" + "github.com/kptdev/porch/controllers/repo-cache" "github.com/kptdev/porch/pkg/cache" cachetypes "github.com/kptdev/porch/pkg/cache/types" "github.com/kptdev/porch/pkg/engine" "github.com/kptdev/porch/pkg/registry/porch" - pkgerrors "github.com/pkg/errors" "google.golang.org/api/option" "google.golang.org/api/sts/v1" corev1 "k8s.io/api/core/v1" @@ -51,12 +49,13 @@ import ( ctrl "sigs.k8s.io/controller-runtime" ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/client" - metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" - "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/manager" ) const NameIndexKey = "metadata.name" +const LeaderElectionID = "porch-server" + var ( // Scheme defines methods for serializing and deserializing API objects. Scheme = runtime.NewScheme() @@ -86,6 +85,13 @@ func init() { ) } +type HAConfig struct { + LeaderElection bool + LeaseDuration time.Duration + RenewDeadline time.Duration + RetryPeriod time.Duration +} + // ExtraConfig holds custom apiserver config type ExtraConfig struct { CoreAPIKubeconfigPath string @@ -93,8 +99,12 @@ type ExtraConfig struct { GRPCRuntimeOptions engine.GRPCRuntimeOptions CacheOptions cachetypes.CacheOptions + HAOptions HAConfig + PodNameSpace string - FunctionStore *reconciler.FunctionConfigStore + FunctionStore *functionconfigs.FunctionConfigStore + + ProbePort int } // Config defines the config for the apiserver @@ -103,17 +113,34 @@ type Config struct { ExtraConfig ExtraConfig } -// PorchServer contains state for a Kubernetes cluster master/api server. -type PorchServer struct { - GenericAPIServer *genericapiserver.GenericAPIServer - coreClient client.WithWatch - cache cachetypes.Cache - ExtraConfig *ExtraConfig +// serverDeps holds injectable construction dependencies for unit testing. +// Production uses defaultServerDeps(); tests override selected fields. +type serverDeps struct { + newManager func(cfg *rest.Config, opts ctrl.Options) (manager.Manager, error) + getCache func(ctx context.Context, opts cachetypes.CacheOptions) (cachetypes.Cache, error) + newSTS func(ctx context.Context, opts ...option.ClientOption) (*sts.Service, error) + newEngine func(opts ...engine.EngineOption) (engine.CaDEngine, error) + // registerFCController, when non-nil, replaces registerFunctionConfigController. + registerFCController func(mgr manager.Manager) error + // registerRCController, when non-nil, replaces registerRepoCacheController. + registerRCController func(mgr manager.Manager) error + cacheRetry wait.Backoff +} + +func defaultServerDeps() serverDeps { + return serverDeps{ + newManager: ctrl.NewManager, + getCache: cache.GetCacheImpl, + newSTS: sts.NewService, + newEngine: engine.NewCaDEngine, + cacheRetry: wait.Backoff{Duration: time.Second, Factor: 1.5, Steps: 20, Cap: 30 * time.Second}, + } } type completedConfig struct { GenericConfig genericapiserver.CompletedConfig ExtraConfig *ExtraConfig + deps serverDeps } // CompletedConfig embeds a private pointer that cannot be instantiated outside of this package. @@ -126,8 +153,9 @@ func (cfg *Config) Complete() CompletedConfig { cfg.GenericConfig.EffectiveVersion = compatibility.NewEffectiveVersionFromString("1.0", "1.0", "1.0") c := completedConfig{ - cfg.GenericConfig.Complete(), - &cfg.ExtraConfig, + GenericConfig: cfg.GenericConfig.Complete(), + ExtraConfig: &cfg.ExtraConfig, + deps: defaultServerDeps(), } return CompletedConfig{&c} @@ -181,124 +209,80 @@ func buildCompleteScheme() (*runtime.Scheme, error) { return completeScheme, err } -func (c completedConfig) getRestConfig() (*rest.Config, error) { +func (c *completedConfig) getRestConfig() (*rest.Config, error) { kubeconfig := c.ExtraConfig.CoreAPIKubeconfigPath if kubeconfig == "" { - icc, err := rest.InClusterConfig() + config, err := rest.InClusterConfig() if err != nil { return nil, fmt.Errorf("failed to load in-cluster config (specify --kubeconfig if not running in-cluster): %w", err) } - return icc, nil - } else { - loadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfig} - loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, &clientcmd.ConfigOverrides{}) - - cc, err := loader.ClientConfig() - if err != nil { - return nil, fmt.Errorf("failed to load config %q: %w", kubeconfig, err) - } - return cc, nil - } -} - -func (c completedConfig) buildClient(ctx context.Context) (client.WithWatch, error) { - restConfig, err := c.getRestConfig() - if err != nil { - return nil, err + return config, nil } - // set high qps/burst limits since this will effectively limit API server responsiveness - restConfig.QPS = 200 - restConfig.Burst = 400 + loadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfig} + loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, &clientcmd.ConfigOverrides{}) - scheme, err := buildCompleteScheme() + config, err := loader.ClientConfig() if err != nil { - return nil, err + return nil, fmt.Errorf("failed to load config %q: %w", kubeconfig, err) } + return config, nil +} - informerCache, err := ctrlcache.New(restConfig, ctrlcache.Options{ +func (c *completedConfig) buildClient(restConfig *rest.Config, scheme *runtime.Scheme, reader client.Reader) (client.WithWatch, error) { + withWatch, err := client.NewWithWatch(restConfig, client.Options{ Scheme: scheme, - ByObject: map[client.Object]ctrlcache.ByObject{ - //The informer should pre-cache all the repositories at startup - &configapi.Repository{}: {}, + Cache: &client.CacheOptions{ + Reader: reader, + DisableFor: []client.Object{ + // The caching client should not cache resources served by porch-server + &porchapi.PackageRevision{}, + &porchapi.PackageRevisionResources{}, + // PackageRev uses write-then-read patterns (Create then Get in + // ClosePackageRevisionDraft/SetMeta). Since writes bypass the + // informer cache, a subsequent Get can miss the just-created object. + &configapi.PackageRev{}, + // v1alpha2 PackageRevision is a CRD patched by patchRenderRequestAnnotation + // right after a write; bypass the cache to avoid stale reads and the + // cluster-scope watch that the informer would require. + &porchv1alpha2.PackageRevision{}, + }, }, }) - - if err != nil { - return nil, fmt.Errorf("failed to create cache: %w", err) - } - - go func() { - if err := informerCache.Start(ctx); err != nil { - klog.Errorf("informer cache stopped with error: %v", err) - } - }() - - if !informerCache.WaitForCacheSync(ctx) { - return nil, fmt.Errorf("informer cache failed to sync") - } - - // Register field index for metadata.name so that field selector queries on Repository work through the cache - if err := informerCache.IndexField(ctx, &configapi.Repository{}, "metadata.name", func(obj client.Object) []string { - return []string{obj.GetName()} - }); err != nil { - return nil, fmt.Errorf("failed to create index for metadata.name: %w", err) - } - - return client.NewWithWatch(restConfig, client.Options{Scheme: scheme, Cache: &client.CacheOptions{ - Reader: informerCache, - DisableFor: []client.Object{ - //The caching client should not cache resources served by porch-server - &porchapi.PackageRevision{}, - &porchapi.PackageRevisionResources{}, - // PackageRev uses write-then-read patterns (Create then Get in - // ClosePackageRevisionDraft/SetMeta). Since writes bypass the - // informer cache, a subsequent Get can miss the just-created object. - // This is not ideal, but crcache doesn't support a level of resources where caching makes a difference - &configapi.PackageRev{}, - // v1alpha2 PackageRevision is a CRD patched by patchRenderRequestAnnotation - // right after a write; bypass the cache to avoid stale reads and the - // cluster-scope watch that the informer would require. - &porchv1alpha2.PackageRevision{}, - }, - }}) -} - -func (c completedConfig) getCoreV1Client() (*corev1client.CoreV1Client, error) { - restConfig, err := c.getRestConfig() - if err != nil { - return nil, err - } - - corev1Client, err := corev1client.NewForConfig(restConfig) if err != nil { - return nil, fmt.Errorf("error building corev1 client: %w", err) + return nil, fmt.Errorf("error building watching client: %w", err) } - return corev1Client, nil + return withWatch, nil } -func (c completedConfig) buildFunctionConfigReconciler(ctx context.Context, scheme *runtime.Scheme, withIndex bool) (*reconciler.FunctionConfigReconciler, error) { - restConfig, err := c.getRestConfig() - if err != nil { - return nil, err - } - - ctx, cancel := context.WithCancel(ctx) - _ = cancel - - var cacheOpts ctrlcache.Options - cacheOpts.Scheme = scheme - //cacheOpts.DefaultNamespaces = map[string]cache.Config{o.exec.PodNamespace: {}} - - mgr, err := ctrl.NewManager(restConfig, ctrl.Options{ - Scheme: scheme, - Cache: cacheOpts, +func (c *completedConfig) buildManager(restConfig *rest.Config, scheme *runtime.Scheme, withIndex bool) (manager.Manager, error) { + probePort := "" + if c.ExtraConfig.ProbePort > 0 { + probePort = fmt.Sprintf(":%d", c.ExtraConfig.ProbePort) + } + + mgr, err := c.deps.newManager(restConfig, ctrl.Options{ + Scheme: scheme, + LeaderElection: c.ExtraConfig.HAOptions.LeaderElection, + LeaderElectionID: LeaderElectionID, + LeaseDuration: zeroToNil(c.ExtraConfig.HAOptions.LeaseDuration), + RenewDeadline: zeroToNil(c.ExtraConfig.HAOptions.RenewDeadline), + RetryPeriod: zeroToNil(c.ExtraConfig.HAOptions.RetryPeriod), + Cache: ctrlcache.Options{ + Scheme: scheme, + ByObject: map[client.Object]ctrlcache.ByObject{ + // The informer should pre-cache all the repositories at startup + &configapi.Repository{}: {}, + }, + }, + HealthProbeBindAddress: probePort, }) if err != nil { - return nil, fmt.Errorf("error at creating manager: %w", err) + return nil, fmt.Errorf("error building manager: %w", err) } if withIndex { + ctx := context.Background() if err := mgr.GetFieldIndexer().IndexField(ctx, &configapi.Repository{}, NameIndexKey, func(o client.Object) []string { repository := o.(*configapi.Repository) @@ -308,67 +292,111 @@ func (c completedConfig) buildFunctionConfigReconciler(ctx context.Context, sche } return []string{repository.Name} }); err != nil { - return nil, err + return nil, fmt.Errorf("error indexing Repository by name: %w", err) } } - functionConfigStore := reconciler.NewFunctionConfigStore(c.ExtraConfig.GRPCRuntimeOptions.DefaultImagePrefix, "") + return mgr, nil +} + +func zeroToNil(duration time.Duration) *time.Duration { + if duration == 0 { + return nil + } + return &duration +} + +func (c *completedConfig) registerFunctionConfigController(mgr manager.Manager) error { + if c.deps.registerFCController != nil { + return c.deps.registerFCController(mgr) + } - rec := &reconciler.FunctionConfigReconciler{ + functionConfigStore := functionconfigs.NewFunctionConfigStore(c.ExtraConfig.GRPCRuntimeOptions.DefaultImagePrefix, "") + + controller := &functionconfigs.Reconciler{ Client: mgr.GetClient(), FunctionConfigStore: functionConfigStore, - For: reconciler.ReconcilerForServer, + For: functionconfigs.ReconcilerForServer, } c.ExtraConfig.FunctionStore = functionConfigStore - err = ctrl.NewControllerManagedBy(mgr). - For(&configapi.FunctionConfig{}). - WithEventFilter(predicate.GenerationChangedPredicate{}). - Complete(rec) - if err != nil { - return nil, fmt.Errorf("error at creating controller: %w", err) + return controller.SetupWithManager(mgr) +} + +func (c *completedConfig) registerRepoCacheController(mgr manager.Manager, cache cachetypes.Cache) error { + if c.deps.registerRCController != nil { + return c.deps.registerRCController(mgr) } + maxConcurrency := c.ExtraConfig.CacheOptions.CRCacheOptions.MaxConcurrentLists - go func() { - if err := mgr.Start(ctx); err != nil { - klog.Infof("manager stopped: %v", err) - } - }() + if maxConcurrency <= 0 { + maxConcurrency = 20 + } + + r := &repocache.Reconciler{ + Client: mgr.GetClient(), + Cache: cache, + MaxConcurrency: maxConcurrency, + } + + return r.SetupWithManager(mgr) +} - return rec, nil +func (c *completedConfig) getCoreV1Client(restConfig *rest.Config) (*corev1client.CoreV1Client, error) { + corev1Client, err := corev1client.NewForConfig(restConfig) + if err != nil { + return nil, fmt.Errorf("error building corev1 client: %w", err) + } + return corev1Client, nil } -// New returns a new instance of PorchServer from the given config. -func (c completedConfig) New(ctx context.Context) (*PorchServer, error) { +// New returns a new manager with PorchServer and FunctionConfigReconciler registered from the given config. +func (c *completedConfig) New(ctx context.Context) (manager.Manager, *PorchServer, error) { // TODO: REMOVE AFTER ASYNC IMPLEMENTATION IS READY. // Set the default request timeout just above hardcoded ctx timeout c.GenericConfig.RequestTimeout = 291 * time.Second genericServer, err := c.GenericConfig.New("porch-apiserver", genericapiserver.NewEmptyDelegate()) if err != nil { - return nil, err + return nil, nil, err + } + + restConfig, err := c.getRestConfig() + if err != nil { + return nil, nil, err } - coreClient, err := c.buildClient(ctx) + // set high qps/burst limits since this will effectively limit API server responsiveness + restConfig.QPS = 200 + restConfig.Burst = 400 + + scheme, err := buildCompleteScheme() if err != nil { - return nil, fmt.Errorf("failed to build client for core apiserver: %w", err) + return nil, nil, fmt.Errorf("error building scheme: %w", err) } - fnConfigReconciler, err := c.buildFunctionConfigReconciler(ctx, coreClient.Scheme(), true) + mgr, err := c.buildManager(restConfig, scheme, true) if err != nil { - return nil, fmt.Errorf("failed to build function config reconciler: %w", err) + return nil, nil, fmt.Errorf("failed to build controller-runtime manager: %w", err) } - c.ExtraConfig.FunctionStore = fnConfigReconciler.FunctionConfigStore + if err = c.registerFunctionConfigController(mgr); err != nil { + return nil, nil, fmt.Errorf("failed to register FunctionConfig controller: %w", err) + } + + coreClient, err := c.buildClient(restConfig, scheme, mgr.GetCache()) + if err != nil { + return nil, nil, fmt.Errorf("failed to build client for core apiserver: %w", err) + } - coreV1Client, err := c.getCoreV1Client() + coreV1Client, err := c.getCoreV1Client(restConfig) if err != nil { - return nil, err + return nil, nil, err } - stsClient, err := sts.NewService(context.Background(), option.WithoutAuthentication()) + stsClient, err := c.deps.newSTS(context.Background(), option.WithoutAuthentication()) if err != nil { - return nil, fmt.Errorf("failed to build sts client: %w", err) + return nil, nil, fmt.Errorf("failed to build sts client: %w", err) } resolverChain := []porch.Resolver{ @@ -393,19 +421,23 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) { var cacheImpl cachetypes.Cache err = retry.OnError( - wait.Backoff{Duration: time.Second, Factor: 1.5, Steps: 20, Cap: 30 * time.Second}, + c.deps.cacheRetry, func(err error) bool { klog.Warningf("failed to create repository cache: %v; wait a sec...", err) return true }, func() error { var err error - cacheImpl, err = cache.GetCacheImpl(ctx, c.ExtraConfig.CacheOptions) + cacheImpl, err = c.deps.getCache(ctx, c.ExtraConfig.CacheOptions) return err }) if err != nil { - return nil, pkgerrors.Wrap(err, "failed to create repository cache") + return nil, nil, fmt.Errorf("failed to create repository cache: %w", err) + } + + if err = c.registerRepoCacheController(mgr, cacheImpl); err != nil { + return nil, nil, fmt.Errorf("failed to setup repo cache controller: %w", err) } runnerOptionsResolver := func(namespace string) runneroptions.RunnerOptions { @@ -414,11 +446,8 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) { return runnerOptions } - cad, err := engine.NewCaDEngine( + cad, err := c.deps.newEngine( engine.WithCache(cacheImpl), - // The order of registering the function runtimes matters here. When - // evaluating a function, the runtimes will be tried in the same - // order as they are registered. engine.WithBuiltinFunctionRuntime(c.ExtraConfig.FunctionStore), engine.WithGRPCFunctionRuntime(c.ExtraConfig.GRPCRuntimeOptions), engine.WithCredentialResolver(credentialResolver), @@ -429,7 +458,7 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) { engine.WithRepoOperationRetryAttempts(c.ExtraConfig.CacheOptions.RepoOperationRetryAttempts), ) if err != nil { - return nil, err + return nil, nil, err } restStorageOptions := porch.RESTStorageOptions{ @@ -440,88 +469,23 @@ func (c completedConfig) New(ctx context.Context) (*PorchServer, error) { } porchGroup, err := restStorageOptions.NewRESTStorage() if err != nil { - return nil, err + return nil, nil, err } - s := &PorchServer{ + porchServer := &PorchServer{ GenericAPIServer: genericServer, coreClient: coreClient, cache: cacheImpl, - ExtraConfig: c.ExtraConfig, - } - - // Install the groups. - if err := s.GenericAPIServer.InstallAPIGroups(&porchGroup); err != nil { - return nil, err - } - - return s, nil -} - -func (s *PorchServer) Run(ctx context.Context) error { - // TODO: Reconsider if the existence of CERT_STORAGE_DIR was a good inidcator for webhook setup, - // but for now we keep backward compatiblity - certStorageDir, found := os.LookupEnv("CERT_STORAGE_DIR") - if found && strings.TrimSpace(certStorageDir) != "" { - if err := setupWebhooks(ctx, s.coreClient); err != nil { - klog.Errorf("%v\n", err) - return err - } - } else { - klog.Infoln("Cert storage dir not provided, skipping webhook setup") - } - - // Start the repo cache eviction controller. Watches for Repository deletions - // and evicts them from the in-memory cache to prevent git clone leaks. - if err := s.startRepoCacheController(ctx); err != nil { - return fmt.Errorf("failed to start repo cache eviction controller: %w", err) - } - - return s.GenericAPIServer.PrepareRun().RunWithContext(ctx) -} - -func (s *PorchServer) startRepoCacheController(ctx context.Context) error { - kubeconfig := s.ExtraConfig.CoreAPIKubeconfigPath - var restConfig *rest.Config - var err error - - if kubeconfig == "" { - restConfig, err = rest.InClusterConfig() - if err != nil { - return fmt.Errorf("failed to load in-cluster config: %w", err) - } - } else { - loadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfig} - loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, &clientcmd.ConfigOverrides{}) - restConfig, err = loader.ClientConfig() - if err != nil { - return fmt.Errorf("failed to load config %q: %w", kubeconfig, err) - } - } - - scheme, err := buildCompleteScheme() - if err != nil { - return err + leaderElect: c.ExtraConfig.HAOptions.LeaderElection, } - mgr, err := ctrl.NewManager(restConfig, ctrl.Options{ - Scheme: scheme, - Metrics: metricsserver.Options{BindAddress: "0"}, // disable metrics - HealthProbeBindAddress: "0", // disable health probes - }) - if err != nil { - return fmt.Errorf("failed to create manager for repo cache controller: %w", err) + if err = porchServer.GenericAPIServer.InstallAPIGroups(&porchGroup); err != nil { + return nil, nil, fmt.Errorf("failed to install Porch API group to PorchServer instance: %w", err) } - if err := setupRepoCacheController(mgr, s.cache, s.ExtraConfig.CacheOptions.CRCacheOptions.MaxConcurrentLists); err != nil { - return fmt.Errorf("failed to setup repo cache controller: %w", err) + if err = mgr.Add(porchServer); err != nil { + return nil, nil, fmt.Errorf("failed to register PorchServer instance to manager: %w", err) } - go func() { - if err := mgr.Start(ctx); err != nil { - klog.Errorf("repo cache controller manager stopped: %v", err) - } - }() - - return nil + return mgr, porchServer, nil } diff --git a/pkg/apiserver/config_test.go b/pkg/apiserver/config_test.go new file mode 100644 index 000000000..088fef909 --- /dev/null +++ b/pkg/apiserver/config_test.go @@ -0,0 +1,147 @@ +// Copyright 2026 The kpt 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 apiserver + +import ( + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + sampleopenapi "github.com/kptdev/porch/api/generated/openapi" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/apiserver/pkg/endpoints/openapi" + genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/client-go/rest" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +func completedConfigForTest(t *testing.T, extra ExtraConfig) CompletedConfig { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { ln.Close() }) + + return (&Config{ + GenericConfig: &genericapiserver.RecommendedConfig{ + Config: genericapiserver.Config{ + SecureServing: &genericapiserver.SecureServingInfo{ + Listener: ln, + }, + }, + }, + ExtraConfig: extra, + }).Complete() +} + +func completedConfigForNewTest(t *testing.T, extra ExtraConfig) CompletedConfig { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { ln.Close() }) + + serverConfig := genericapiserver.NewRecommendedConfig(Codecs) + serverConfig.SecureServing = &genericapiserver.SecureServingInfo{Listener: ln} + serverConfig.LoopbackClientConfig = &rest.Config{Host: "https://127.0.0.1:1"} + serverConfig.OpenAPIConfig = genericapiserver.DefaultOpenAPIConfig(sampleopenapi.GetOpenAPIDefinitions, openapi.NewDefinitionNamer(Scheme)) + serverConfig.OpenAPIV3Config = genericapiserver.DefaultOpenAPIV3Config(sampleopenapi.GetOpenAPIDefinitions, openapi.NewDefinitionNamer(Scheme)) + + completed := (&Config{ + GenericConfig: serverConfig, + ExtraConfig: extra, + }).Complete() + return completed +} + +func restConfigWithFakeAPIServer(t *testing.T) *rest.Config { + t.Helper() + + mux := http.NewServeMux() + mux.HandleFunc("/api", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"kind":"APIVersions","versions":["v1"]}`)) + }) + mux.HandleFunc("/apis", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"kind":"APIGroupList","groups":[{"name":"config.porch.kpt.dev","versions":[{"groupVersion":"config.porch.kpt.dev/v1alpha1","version":"v1alpha1"}]}]}`)) + }) + mux.HandleFunc("/apis/config.porch.kpt.dev/v1alpha1", func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{ + "kind":"APIResourceList", + "apiVersion":"config.porch.kpt.dev/v1alpha1", + "resources":[ + {"name":"repositories","singularName":"repository","namespaced":true,"kind":"Repository","verbs":["get","list","watch"]}, + {"name":"functionconfigs","singularName":"functionconfig","namespaced":true,"kind":"FunctionConfig","verbs":["get","list","watch"]} + ] + }`)) + }) + + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return &rest.Config{Host: srv.URL} +} + +func TestBuildManager(t *testing.T) { + scheme, err := buildCompleteScheme() + require.NoError(t, err) + + restConfig := restConfigWithFakeAPIServer(t) + + t.Run("without repository index", func(t *testing.T) { + completed := completedConfigForTest(t, ExtraConfig{}) + mgr, err := completed.buildManager(restConfig, scheme, false) + require.NoError(t, err) + assert.NotNil(t, mgr) + }) + + t.Run("with repository index and probe port", func(t *testing.T) { + completed := completedConfigForTest(t, ExtraConfig{ + ProbePort: 4453, + HAOptions: HAConfig{ + LeaseDuration: 15 * time.Second, + }, + }) + completed.deps.newManager = func(cfg *rest.Config, opts ctrl.Options) (manager.Manager, error) { + opts.HealthProbeBindAddress = "0" + return ctrl.NewManager(cfg, opts) + } + + mgr, err := completed.buildManager(restConfig, scheme, true) + require.NoError(t, err) + assert.NotNil(t, mgr) + }) +} + +func TestRegisterFunctionConfigController(t *testing.T) { + scheme, err := buildCompleteScheme() + require.NoError(t, err) + + restConfig := restConfigWithFakeAPIServer(t) + completed := completedConfigForTest(t, ExtraConfig{}) + + mgr, err := completed.buildManager(restConfig, scheme, false) + require.NoError(t, err) + + err = completed.registerFunctionConfigController(mgr) + require.NoError(t, err) + assert.NotNil(t, completed.ExtraConfig.FunctionStore) +} + +func TestLeaderElectionID(t *testing.T) { + assert.Equal(t, "porch-server", LeaderElectionID) +} diff --git a/pkg/apiserver/new_test.go b/pkg/apiserver/new_test.go new file mode 100644 index 000000000..3669512e8 --- /dev/null +++ b/pkg/apiserver/new_test.go @@ -0,0 +1,189 @@ +// Copyright 2026 The kpt 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 apiserver + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/kptdev/porch/controllers/functionconfigs" + cachetypes "github.com/kptdev/porch/pkg/cache/types" + "github.com/kptdev/porch/pkg/engine" + mockcachetypes "github.com/kptdev/porch/test/mockery/mocks/porch/pkg/cache/types" + mockengine "github.com/kptdev/porch/test/mockery/mocks/porch/pkg/engine" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/rest" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +func TestBuildManagerPassesProbePort(t *testing.T) { + scheme, err := buildCompleteScheme() + require.NoError(t, err) + + var gotOpts ctrl.Options + completed := completedConfigForTest(t, ExtraConfig{ProbePort: 4453}) + completed.deps.newManager = func(cfg *rest.Config, opts ctrl.Options) (manager.Manager, error) { + gotOpts = opts + opts.HealthProbeBindAddress = "0" + return ctrl.NewManager(cfg, opts) + } + + mgr, err := completed.buildManager(restConfigWithFakeAPIServer(t), scheme, false) + require.NoError(t, err) + assert.NotNil(t, mgr) + assert.Equal(t, ":4453", gotOpts.HealthProbeBindAddress) + assert.Equal(t, LeaderElectionID, gotOpts.LeaderElectionID) +} + +func TestNewWithInjectedDeps(t *testing.T) { + restCfg := restConfigWithFakeAPIServer(t) + kubeconfig := writeTempKubeconfig(t, restCfg.Host) + + completed := completedConfigForNewTest(t, ExtraConfig{ + CoreAPIKubeconfigPath: kubeconfig, + HAOptions: HAConfig{LeaderElection: true}, + CacheOptions: cachetypes.CacheOptions{CacheType: cachetypes.CRCacheType}, + }) + + fakeCache := mockcachetypes.NewMockCache(t) + fakeEngine := mockengine.NewMockCaDEngine(t) + + completed.deps.getCache = func(ctx context.Context, opts cachetypes.CacheOptions) (cachetypes.Cache, error) { + assert.NotNil(t, opts.CoreClient) + return fakeCache, nil + } + completed.deps.newEngine = func(opts ...engine.EngineOption) (engine.CaDEngine, error) { + return fakeEngine, nil + } + completed.deps.registerFCController = func(mgr manager.Manager) error { + completed.ExtraConfig.FunctionStore = functionconfigs.NewFunctionConfigStore("prefix/", "") + return nil + } + completed.deps.registerRCController = func(manager.Manager) error { + return nil + } + completed.deps.newManager = func(cfg *rest.Config, opts ctrl.Options) (manager.Manager, error) { + // Avoid in-cluster leader-election namespace requirement in unit tests. + opts.LeaderElection = false + opts.HealthProbeBindAddress = "0" + return ctrl.NewManager(cfg, opts) + } + completed.deps.cacheRetry = wait.Backoff{Steps: 1} + + mgr, server, err := completed.New(context.Background()) + require.NoError(t, err) + require.NotNil(t, mgr) + require.NotNil(t, server) + assert.True(t, server.NeedLeaderElection()) + assert.Same(t, fakeCache, server.cache) +} + +func TestNewManagerError(t *testing.T) { + restCfg := restConfigWithFakeAPIServer(t) + kubeconfig := writeTempKubeconfig(t, restCfg.Host) + + completed := completedConfigForNewTest(t, ExtraConfig{ + CoreAPIKubeconfigPath: kubeconfig, + }) + completed.deps.newManager = func(cfg *rest.Config, opts ctrl.Options) (manager.Manager, error) { + return nil, fmt.Errorf("manager boom") + } + + _, server, err := completed.New(context.Background()) + assert.Nil(t, server) + assert.ErrorContains(t, err, "failed to build controller-runtime manager") + assert.ErrorContains(t, err, "manager boom") +} + +func TestNewCacheError(t *testing.T) { + restCfg := restConfigWithFakeAPIServer(t) + kubeconfig := writeTempKubeconfig(t, restCfg.Host) + + completed := completedConfigForNewTest(t, ExtraConfig{ + CoreAPIKubeconfigPath: kubeconfig, + }) + completed.deps.registerFCController = func(mgr manager.Manager) error { + completed.ExtraConfig.FunctionStore = functionconfigs.NewFunctionConfigStore("prefix/", "") + return nil + } + completed.deps.registerRCController = func(manager.Manager) error { + return nil + } + completed.deps.getCache = func(ctx context.Context, opts cachetypes.CacheOptions) (cachetypes.Cache, error) { + return nil, fmt.Errorf("cache boom") + } + completed.deps.cacheRetry = wait.Backoff{Steps: 1} + + _, server, err := completed.New(context.Background()) + assert.Nil(t, server) + assert.ErrorContains(t, err, "failed to create repository cache") + assert.ErrorContains(t, err, "cache boom") +} + +func TestNewEngineError(t *testing.T) { + restCfg := restConfigWithFakeAPIServer(t) + kubeconfig := writeTempKubeconfig(t, restCfg.Host) + + completed := completedConfigForNewTest(t, ExtraConfig{ + CoreAPIKubeconfigPath: kubeconfig, + }) + completed.deps.registerFCController = func(mgr manager.Manager) error { + completed.ExtraConfig.FunctionStore = functionconfigs.NewFunctionConfigStore("prefix/", "") + return nil + } + completed.deps.registerRCController = func(manager.Manager) error { + return nil + } + completed.deps.getCache = func(ctx context.Context, opts cachetypes.CacheOptions) (cachetypes.Cache, error) { + return mockcachetypes.NewMockCache(t), nil + } + completed.deps.newEngine = func(opts ...engine.EngineOption) (engine.CaDEngine, error) { + return nil, fmt.Errorf("engine boom") + } + completed.deps.cacheRetry = wait.Backoff{Steps: 1} + + _, server, err := completed.New(context.Background()) + assert.Nil(t, server) + assert.ErrorContains(t, err, "engine boom") +} + +func writeTempKubeconfig(t *testing.T, server string) string { + t.Helper() + content := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- cluster: + server: %s + name: test +contexts: +- context: + cluster: test + user: test + name: test +current-context: test +users: +- name: test + user: {} +`, server) + path := filepath.Join(t.TempDir(), "kubeconfig") + require.NoError(t, os.WriteFile(path, []byte(content), 0o600)) + return path +} diff --git a/pkg/apiserver/porchserver.go b/pkg/apiserver/porchserver.go new file mode 100644 index 000000000..388d4b981 --- /dev/null +++ b/pkg/apiserver/porchserver.go @@ -0,0 +1,58 @@ +// Copyright 2026 The kpt 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 apiserver + +import ( + "context" + "os" + "strings" + + cachetypes "github.com/kptdev/porch/pkg/cache/types" + genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/klog/v2" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/manager" +) + +// PorchServer contains state for a Kubernetes cluster master/api server. +type PorchServer struct { + GenericAPIServer *genericapiserver.GenericAPIServer + + leaderElect bool + coreClient client.WithWatch + cache cachetypes.Cache +} + +var _ manager.Runnable = &PorchServer{} +var _ manager.LeaderElectionRunnable = &PorchServer{} + +func (s *PorchServer) Start(ctx context.Context) error { + // TODO: Reconsider if the existence of CERT_STORAGE_DIR was a good indicator for webhook setup, + // but for now we keep backward compatibility + certStorageDir, found := os.LookupEnv("CERT_STORAGE_DIR") + if found && strings.TrimSpace(certStorageDir) != "" { + if err := setupWebhooks(ctx, s.coreClient); err != nil { + klog.Errorf("%v\n", err) + return err + } + } else { + klog.Infoln("Cert storage dir not provided, skipping webhook setup") + } + return s.GenericAPIServer.PrepareRun().RunWithContext(ctx) +} + +func (s *PorchServer) NeedLeaderElection() bool { + return s.leaderElect +} diff --git a/pkg/cmd/server/start.go b/pkg/cmd/server/start.go index 0f51c6b9c..b55da9b3f 100644 --- a/pkg/cmd/server/start.go +++ b/pkg/cmd/server/start.go @@ -19,6 +19,7 @@ import ( "fmt" "io" "net" + "net/http" "os" "strings" "time" @@ -26,6 +27,7 @@ import ( "github.com/kptdev/kpt/pkg/lib/runneroptions" "github.com/spf13/cobra" "github.com/spf13/pflag" + "sigs.k8s.io/controller-runtime/pkg/healthz" clientset "github.com/kptdev/porch/api/generated/clientset/versioned" informers "github.com/kptdev/porch/api/generated/informers/externalversions" @@ -43,6 +45,7 @@ import ( genericapiserver "k8s.io/apiserver/pkg/server" genericoptions "k8s.io/apiserver/pkg/server/options" utilfeature "k8s.io/apiserver/pkg/util/feature" + "k8s.io/client-go/rest" "k8s.io/klog/v2" netutils "k8s.io/utils/net" ) @@ -89,6 +92,10 @@ type PorchServerOptions struct { UseUserDefinedCaBundle bool PodNamespace string + + ProbePort int + + HAOptions apiserver.HAConfig } // NewPorchServerOptions returns a new PorchServerOptions @@ -321,41 +328,46 @@ func (o *PorchServerOptions) Config() (*apiserver.Config, error) { return nil, err } - config := &apiserver.Config{ + return &apiserver.Config{ GenericConfig: serverConfig, - ExtraConfig: apiserver.ExtraConfig{ - CoreAPIKubeconfigPath: o.CoreAPIKubeconfigPath, - GRPCRuntimeOptions: engine.GRPCRuntimeOptions{ - FunctionRunnerAddress: o.FunctionRunnerAddress, - MaxGrpcMessageSize: o.MaxRequestBodySize, - DefaultImagePrefix: o.DefaultImagePrefix, + ExtraConfig: o.buildExtraConfig(), + }, nil +} + +func (o *PorchServerOptions) buildExtraConfig() apiserver.ExtraConfig { + return apiserver.ExtraConfig{ + CoreAPIKubeconfigPath: o.CoreAPIKubeconfigPath, + GRPCRuntimeOptions: engine.GRPCRuntimeOptions{ + FunctionRunnerAddress: o.FunctionRunnerAddress, + MaxGrpcMessageSize: o.MaxRequestBodySize, + DefaultImagePrefix: o.DefaultImagePrefix, + }, + CacheOptions: cachetypes.CacheOptions{ + ExternalRepoOptions: externalrepotypes.ExternalRepoOptions{ + LocalDirectory: o.CacheDirectory, + UseUserDefinedCaBundle: o.UseUserDefinedCaBundle, + GoGitRepoCacheSize: o.GoGitRepoCacheSize, + GoGitCacheMaxFileSize: o.GoGitCacheMaxFileSize, + }, + RepoOperationRetryAttempts: o.RepoOperationRetryAttempts, + CacheType: cachetypes.CacheType(o.CacheType), + CRCacheOptions: cachetypes.CRCacheOptions{ + MaxConcurrentLists: o.MaxConcurrentLists, + ListTimeoutPerRepository: o.ListTimeoutPerRepository, }, - CacheOptions: cachetypes.CacheOptions{ - ExternalRepoOptions: externalrepotypes.ExternalRepoOptions{ - LocalDirectory: o.CacheDirectory, - UseUserDefinedCaBundle: o.UseUserDefinedCaBundle, - GoGitRepoCacheSize: o.GoGitRepoCacheSize, - GoGitCacheMaxFileSize: o.GoGitCacheMaxFileSize, - }, - RepoOperationRetryAttempts: o.RepoOperationRetryAttempts, - CacheType: cachetypes.CacheType(o.CacheType), - CRCacheOptions: cachetypes.CRCacheOptions{ - MaxConcurrentLists: o.MaxConcurrentLists, - ListTimeoutPerRepository: o.ListTimeoutPerRepository, - }, - DBCacheOptions: cachetypes.DBCacheOptions{ - Driver: o.DbCacheDriver, - DataSource: o.DbCacheDataSource, - MaxConnections: o.DbMaxConnections, - MaxIdleConnections: o.DbMaxIdleConnections, - MaxConnLifetime: o.DbMaxConnLifetime, - }, - DbPushDraftsToGit: o.DbPushDrafsToGit, + DBCacheOptions: cachetypes.DBCacheOptions{ + Driver: o.DbCacheDriver, + DataSource: o.DbCacheDataSource, + MaxConnections: o.DbMaxConnections, + MaxIdleConnections: o.DbMaxIdleConnections, + MaxConnLifetime: o.DbMaxConnLifetime, }, - PodNameSpace: o.PodNamespace, + DbPushDraftsToGit: o.DbPushDrafsToGit, }, + PodNameSpace: o.PodNamespace, + ProbePort: o.ProbePort, + HAOptions: o.HAOptions, } - return config, nil } // RunPorchServer starts a new PorchServer given PorchServerOptions @@ -365,7 +377,7 @@ func (o PorchServerOptions) RunPorchServer(ctx context.Context) error { return err } - server, err := config.Complete().New(ctx) + mgr, server, err := config.Complete().New(ctx) if err != nil { return err } @@ -378,7 +390,76 @@ func (o PorchServerOptions) RunPorchServer(ctx context.Context) error { }) } - return server.Run(ctx) + if o.ProbePort > 0 { + loopback := config.GenericConfig.LoopbackClientConfig + if loopback == nil { + return fmt.Errorf("loopback client config is required to proxy health checks") + } + client, err := rest.HTTPClientFor(loopback) + if err != nil { + return fmt.Errorf("failed to create loopback health client: %w", err) + } + client.Timeout = 3 * time.Second + if err := proxyHealthChecks(mgr, client, loopback.Host); err != nil { + return err + } + } + + return mgr.Start(ctx) +} + +type probeManager interface { + AddHealthzCheck(name string, check healthz.Checker) error + AddReadyzCheck(name string, check healthz.Checker) error + Elected() <-chan struct{} +} + +func proxyHealthChecks(mgr probeManager, client *http.Client, baseURL string) error { + healthzDelegate := delegateAPIServerHealth(mgr, client, baseURL, "healthz", true) + if err := mgr.AddHealthzCheck("healthz", healthzDelegate); err != nil { + return err + } + + // there is no livez on the controller-runtime manager, only healthz, so we proxy both to healthz + livezDelegate := delegateAPIServerHealth(mgr, client, baseURL, "livez", true) + if err := mgr.AddHealthzCheck("livez", livezDelegate); err != nil { + return err + } + + readyzDelegate := delegateAPIServerHealth(mgr, client, baseURL, "readyz", false) + if err := mgr.AddReadyzCheck("readyz", readyzDelegate); err != nil { + return err + } + + return nil +} + +type electedManager interface { + Elected() <-chan struct{} +} + +func delegateAPIServerHealth(mgr electedManager, client *http.Client, baseURL, path string, okWhenStandby bool) healthz.Checker { + url := strings.TrimRight(baseURL, "/") + "/" + path + + return func(*http.Request) error { + select { + case <-mgr.Elected(): + resp, err := client.Get(url) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("apiserver %s returned %d", path, resp.StatusCode) + } + return nil + default: + if okWhenStandby { + return nil + } + return fmt.Errorf("not leader") + } + } } func (o *PorchServerOptions) AddFlags(fs *pflag.FlagSet) { @@ -416,4 +497,11 @@ func (o *PorchServerOptions) AddFlags(fs *pflag.FlagSet) { fs.StringSliceVar(&o.RetryableGitErrors, "retryable-git-errors", nil, "Additional retryable git error patterns. Can be specified multiple times or as comma-separated values.") fs.DurationVar(&o.ListTimeoutPerRepository, "list-timeout-per-repo", 20*time.Second, "Maximum amount of time to wait for a repository list request.") fs.IntVar(&o.MaxConcurrentLists, "max-parallel-repo-lists", 10, "Maximum number of repositories to list in parallel.") + + fs.IntVar(&o.ProbePort, "probe-port", 0, "If > 0, start serving controller-runtime /healthz and /readyz on this port (in addition to the API server's built-in probes at `--secure-port`); a liveness-style check is available as /healthz/livez") + + fs.BoolVar(&o.HAOptions.LeaderElection, "leader-elect", false, "If true, the porch-server will attempt to acquire leader election lock") + fs.DurationVar(&o.HAOptions.LeaseDuration, "leader-lease-duration", 0, "The duration that non-leader candidates will wait to force acquire leadership") + fs.DurationVar(&o.HAOptions.RenewDeadline, "leader-renew-deadline", 0, "The duration that the acting controlplane will retry refreshing leadership before giving up") + fs.DurationVar(&o.HAOptions.RetryPeriod, "leader-retry-period", 0, "The duration the LeaderElector clients should wait between tries of actions") } diff --git a/pkg/cmd/server/start_test.go b/pkg/cmd/server/start_test.go index 9e55f6261..928737b77 100644 --- a/pkg/cmd/server/start_test.go +++ b/pkg/cmd/server/start_test.go @@ -15,16 +15,26 @@ package server import ( + "context" + "fmt" + "net" + "net/http" + "net/http/httptest" "os" + "path/filepath" + "strings" "testing" "time" porchapi "github.com/kptdev/porch/api/porch/v1alpha1" "github.com/kptdev/porch/pkg/apiserver" + cachetypes "github.com/kptdev/porch/pkg/cache/types" "github.com/spf13/pflag" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/runtime/schema" genericoptions "k8s.io/apiserver/pkg/server/options" + "sigs.k8s.io/controller-runtime/pkg/healthz" ) func TestAddFlags(t *testing.T) { @@ -66,17 +76,6 @@ func TestValidate(t *testing.T) { assert.Error(t, err) } -func TestComplete(t *testing.T) { - opts := NewPorchServerOptions(os.Stdout, os.Stderr) - - // Test with patterns that have leading/trailing spaces - opts.RetryableGitErrors = []string{" git error 1 ", "git error 2", " git error 3"} - assert.Len(t, opts.RetryableGitErrors, 3) - - err := opts.Complete() - assert.NoError(t, err) -} - func TestSetupDBCacheConn(t *testing.T) { // Default expected connection pooling values defaultMaxConns := 300 @@ -225,3 +224,272 @@ func TestSetupDBCacheConn(t *testing.T) { }) } } + +func TestComplete(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + + opts.RetryableGitErrors = []string{" git error 1 ", "git error 2", " git error 3"} + assert.Len(t, opts.RetryableGitErrors, 3) + + err := opts.Complete() + assert.NoError(t, err) +} + +func TestHAFlagParsing(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + fs := pflag.NewFlagSet("test", pflag.ContinueOnError) + opts.AddFlags(fs) + require.NoError(t, fs.Parse([]string{ + "--probe-port=4453", + "--leader-elect=true", + "--leader-lease-duration=15s", + })) + assert.Equal(t, 4453, opts.ProbePort) + assert.True(t, opts.HAOptions.LeaderElection) + assert.Equal(t, 15*time.Second, opts.HAOptions.LeaseDuration) +} + +func TestDelegateAPIServerHealthStandby(t *testing.T) { + mgr := &stubProbeManager{elected: make(chan struct{})} + client := &http.Client{} + + liveness := delegateAPIServerHealth(mgr, client, "https://example.invalid", "livez", true) + assert.NoError(t, liveness(nil)) + + readiness := delegateAPIServerHealth(mgr, client, "https://example.invalid", "readyz", false) + assert.ErrorContains(t, readiness(nil), "not leader") +} + +func TestValidateMaxConcurrentLists(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + opts.CacheType = "CR" + opts.MaxConcurrentLists = -1 + + err := opts.Validate(nil) + assert.ErrorContains(t, err, "invalid value for max-parallel-repo-lists") +} + +func TestCompleteSetsDefaultCacheDirectory(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + opts.CacheDirectory = "" + + require.NoError(t, opts.Complete()) + assert.NotEmpty(t, opts.CacheDirectory) + assert.True(t, strings.HasSuffix(opts.CacheDirectory, "/porch")) +} + +func TestCompleteUppercasesCacheType(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + opts.CacheType = "cr" + + require.NoError(t, opts.Complete()) + assert.Equal(t, "CR", opts.CacheType) +} + +func TestNewPorchServerOptionsDefaults(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + + assert.Nil(t, opts.RecommendedOptions.Etcd) + assert.Equal(t, 0, opts.ProbePort) + assert.False(t, opts.HAOptions.LeaderElection) +} + +func TestBuildExtraConfig(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + opts.CoreAPIKubeconfigPath = "/tmp/kubeconfig" + opts.FunctionRunnerAddress = "function-runner:9445" + opts.MaxRequestBodySize = 1234 + opts.DefaultImagePrefix = "example.com/" + opts.CacheDirectory = "/cache" + opts.UseUserDefinedCaBundle = true + opts.GoGitRepoCacheSize = 16 + opts.GoGitCacheMaxFileSize = 1024 + opts.RepoOperationRetryAttempts = 5 + opts.CacheType = "DB" + opts.MaxConcurrentLists = 7 + opts.ListTimeoutPerRepository = 30 * time.Second + opts.DbCacheDriver = "pgx" + opts.DbCacheDataSource = "postgres://x" + opts.DbMaxConnections = 10 + opts.DbMaxIdleConnections = 5 + opts.DbMaxConnLifetime = time.Minute + opts.DbPushDrafsToGit = true + opts.PodNamespace = "test-ns" + opts.ProbePort = 4453 + opts.HAOptions = apiserver.HAConfig{LeaderElection: true, LeaseDuration: 15 * time.Second} + + extra := opts.buildExtraConfig() + + assert.Equal(t, "/tmp/kubeconfig", extra.CoreAPIKubeconfigPath) + assert.Equal(t, "function-runner:9445", extra.GRPCRuntimeOptions.FunctionRunnerAddress) + assert.Equal(t, 1234, extra.GRPCRuntimeOptions.MaxGrpcMessageSize) + assert.Equal(t, "example.com/", extra.GRPCRuntimeOptions.DefaultImagePrefix) + assert.Equal(t, "/cache", extra.CacheOptions.ExternalRepoOptions.LocalDirectory) + assert.True(t, extra.CacheOptions.ExternalRepoOptions.UseUserDefinedCaBundle) + assert.Equal(t, 16, extra.CacheOptions.ExternalRepoOptions.GoGitRepoCacheSize) + assert.Equal(t, int64(1024), extra.CacheOptions.ExternalRepoOptions.GoGitCacheMaxFileSize) + assert.Equal(t, 5, extra.CacheOptions.RepoOperationRetryAttempts) + assert.Equal(t, cachetypes.CacheType("DB"), extra.CacheOptions.CacheType) + assert.Equal(t, 7, extra.CacheOptions.CRCacheOptions.MaxConcurrentLists) + assert.Equal(t, 30*time.Second, extra.CacheOptions.CRCacheOptions.ListTimeoutPerRepository) + assert.Equal(t, "pgx", extra.CacheOptions.DBCacheOptions.Driver) + assert.Equal(t, "postgres://x", extra.CacheOptions.DBCacheOptions.DataSource) + assert.Equal(t, 10, extra.CacheOptions.DBCacheOptions.MaxConnections) + assert.Equal(t, 5, extra.CacheOptions.DBCacheOptions.MaxIdleConnections) + assert.Equal(t, time.Minute, extra.CacheOptions.DBCacheOptions.MaxConnLifetime) + assert.True(t, extra.CacheOptions.DbPushDraftsToGit) + assert.Equal(t, "test-ns", extra.PodNameSpace) + assert.Equal(t, 4453, extra.ProbePort) + assert.True(t, extra.HAOptions.LeaderElection) + assert.Equal(t, 15*time.Second, extra.HAOptions.LeaseDuration) +} + +func TestConfigMapsExtraConfig(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { ln.Close() }) + + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + opts.RecommendedOptions.SecureServing.Listener = ln + opts.RecommendedOptions.SecureServing.ServerCert.CertDirectory = t.TempDir() + opts.LocalStandaloneDebugging = true + opts.CacheType = "CR" + opts.ProbePort = 4453 + opts.HAOptions = apiserver.HAConfig{LeaderElection: true} + opts.PodNamespace = "test-ns" + opts.MaxRequestBodySize = 1234 + opts.CoreAPIKubeconfigPath = writeTempKubeconfig(t, "http://127.0.0.1:1") + opts.RecommendedOptions.CoreAPI.CoreAPIKubeconfigPath = opts.CoreAPIKubeconfigPath + + require.NoError(t, opts.Complete()) + + config, err := opts.Config() + require.NoError(t, err) + assert.Equal(t, 4453, config.ExtraConfig.ProbePort) + assert.True(t, config.ExtraConfig.HAOptions.LeaderElection) + assert.Equal(t, "test-ns", config.ExtraConfig.PodNameSpace) + assert.Equal(t, 1234, config.ExtraConfig.GRPCRuntimeOptions.MaxGrpcMessageSize) + assert.Equal(t, int64(1234), config.GenericConfig.MaxRequestBodyBytes) +} + +func writeTempKubeconfig(t *testing.T, server string) string { + t.Helper() + content := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- cluster: + server: %s + name: test +contexts: +- context: + cluster: test + user: test + name: test +current-context: test +users: +- name: test + user: {} +`, server) + path := filepath.Join(t.TempDir(), "kubeconfig") + require.NoError(t, os.WriteFile(path, []byte(content), 0o600)) + return path +} + +func TestProxyHealthChecks(t *testing.T) { + mgr := &stubProbeManager{elected: make(chan struct{})} + require.NoError(t, proxyHealthChecks(mgr, &http.Client{}, "https://example.invalid")) + assert.Equal(t, []string{"healthz", "livez"}, mgr.healthz) + assert.Equal(t, []string{"readyz"}, mgr.readyz) +} + +func TestProxyHealthChecksErrors(t *testing.T) { + t.Run("healthz registration fails", func(t *testing.T) { + mgr := &stubProbeManager{ + elected: make(chan struct{}), + failHealth: true, + } + assert.ErrorContains(t, proxyHealthChecks(mgr, &http.Client{}, "https://example.invalid"), "healthz error") + }) + t.Run("readyz registration fails", func(t *testing.T) { + mgr := &stubProbeManager{ + elected: make(chan struct{}), + failReady: true, + } + assert.ErrorContains(t, proxyHealthChecks(mgr, &http.Client{}, "https://example.invalid"), "readyz error") + }) +} + +func TestDelegateAPIServerHealthLeader(t *testing.T) { + tests := map[string]struct { + statusCode int + path string + expectedErr string + }{ + "healthy": { + statusCode: http.StatusOK, + path: "healthz", + }, + "not ready": { + statusCode: http.StatusServiceUnavailable, + path: "readyz", + expectedErr: "apiserver readyz returned 503", + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(tc.statusCode) + })) + t.Cleanup(srv.Close) + + mgr := &stubProbeManager{elected: make(chan struct{})} + close(mgr.elected) + + // srv.Client() trusts the test server certificate (same pattern as LoopbackClientConfig). + checker := delegateAPIServerHealth(mgr, srv.Client(), srv.URL, tc.path, true) + err := checker(nil) + if tc.expectedErr == "" { + assert.NoError(t, err) + } else { + assert.ErrorContains(t, err, tc.expectedErr) + } + }) + } +} + +func TestNewCommandStartPorchServer(t *testing.T) { + opts := NewPorchServerOptions(os.Stdout, os.Stderr) + cmd := NewCommandStartPorchServer(context.Background(), opts) + + assert.Equal(t, "Launch a porch API server", cmd.Short) + assert.NotNil(t, cmd.Flags().Lookup("cache-type")) +} + +type stubProbeManager struct { + elected chan struct{} + healthz []string + readyz []string + failHealth bool + failReady bool +} + +func (m *stubProbeManager) Elected() <-chan struct{} { + return m.elected +} + +func (m *stubProbeManager) AddHealthzCheck(name string, _ healthz.Checker) error { + if m.failHealth { + return fmt.Errorf("healthz error") + } + m.healthz = append(m.healthz, name) + return nil +} + +func (m *stubProbeManager) AddReadyzCheck(name string, _ healthz.Checker) error { + if m.failReady { + return fmt.Errorf("readyz error") + } + m.readyz = append(m.readyz, name) + return nil +} diff --git a/pkg/engine/builtinruntime.go b/pkg/engine/builtinruntime.go index da3b3e731..4555719e1 100644 --- a/pkg/engine/builtinruntime.go +++ b/pkg/engine/builtinruntime.go @@ -24,17 +24,17 @@ import ( "github.com/kptdev/kpt/pkg/fn" "github.com/kptdev/kpt/pkg/lib/kptops" fnsdk "github.com/kptdev/krm-functions-sdk/go/fn" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" imageutil "github.com/kptdev/porch/pkg/util/image" regclientref "github.com/regclient/regclient/types/ref" "k8s.io/klog/v2" ) type builtinRuntime struct { - store *reconciler.FunctionConfigStore + store *functionconfigs.FunctionConfigStore } -func newBuiltinRuntime(functionConfigStore *reconciler.FunctionConfigStore) *builtinRuntime { +func newBuiltinRuntime(functionConfigStore *functionconfigs.FunctionConfigStore) *builtinRuntime { return &builtinRuntime{ store: functionConfigStore, } diff --git a/pkg/engine/builtinruntime_test.go b/pkg/engine/builtinruntime_test.go index 15b14fc37..989b9dad4 100644 --- a/pkg/engine/builtinruntime_test.go +++ b/pkg/engine/builtinruntime_test.go @@ -25,7 +25,7 @@ import ( "github.com/kptdev/kpt/pkg/fn" fnsdk "github.com/kptdev/krm-functions-sdk/go/fn" configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" imageutil "github.com/kptdev/porch/pkg/util/image" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -59,7 +59,7 @@ func TestNewBuiltinRuntime(t *testing.T) { }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(applyReplacementsFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) @@ -94,7 +94,7 @@ func TestNewBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(applyReplacementsFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) @@ -132,7 +132,7 @@ func TestBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) funct := &kptfilev1.Function{ @@ -158,7 +158,7 @@ func TestBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) funct := &kptfilev1.Function{ @@ -184,7 +184,7 @@ func TestBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) funct := &kptfilev1.Function{ @@ -211,7 +211,7 @@ func TestBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) funct := &kptfilev1.Function{ @@ -236,7 +236,7 @@ func TestBuiltinRuntime(t *testing.T) { }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(applyReplacementsFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) fn := &kptfilev1.Function{ @@ -279,7 +279,7 @@ functionConfig: }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) fn := &kptfilev1.Function{ @@ -351,7 +351,7 @@ functionConfig: }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) fn := &kptfilev1.Function{ @@ -417,7 +417,7 @@ functionConfig: }, }, } - functionConfigStore := reconciler.NewFunctionConfigStore(defaultKRMImagePrefix, "") + functionConfigStore := functionconfigs.NewFunctionConfigStore(defaultKRMImagePrefix, "") functionConfigStore.UpdateExecCache(setNamespaceFunction, &functionConfig) br := newBuiltinRuntime(functionConfigStore) fn := &kptfilev1.Function{ diff --git a/pkg/engine/grpcruntime.go b/pkg/engine/grpcruntime.go index d54163622..0404d4cfe 100644 --- a/pkg/engine/grpcruntime.go +++ b/pkg/engine/grpcruntime.go @@ -22,7 +22,7 @@ import ( kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" "github.com/kptdev/kpt/pkg/fn" "github.com/kptdev/kpt/pkg/lib/kptops" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" "github.com/kptdev/porch/func/evaluator" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "google.golang.org/grpc" @@ -120,7 +120,7 @@ func (gr *grpcRunner) Run(r io.Reader, w io.Writer) error { // NewMultiFunctionRuntime creates a FunctionRuntime that tries builtin functions // first, then falls back to the gRPC fn-runner. -func NewMultiFunctionRuntime(grpcAddress string, maxGrpcMessageSize int, functionConfigStore *reconciler.FunctionConfigStore) (fn.FunctionRuntime, error) { +func NewMultiFunctionRuntime(grpcAddress string, maxGrpcMessageSize int, functionConfigStore *functionconfigs.FunctionConfigStore) (fn.FunctionRuntime, error) { builtin := newBuiltinRuntime(functionConfigStore) if grpcAddress == "" { diff --git a/pkg/engine/grpcruntime_test.go b/pkg/engine/grpcruntime_test.go index 30980ae8f..e8eb34aa0 100644 --- a/pkg/engine/grpcruntime_test.go +++ b/pkg/engine/grpcruntime_test.go @@ -24,7 +24,7 @@ import ( "testing" v1 "github.com/kptdev/kpt/api/kptfile/v1" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" "github.com/kptdev/porch/func/evaluator" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -350,6 +350,6 @@ func TestNewMultiFunctionRuntime_NilStorePanics(t *testing.T) { }) } -func newTestFunctionConfigStore() *reconciler.FunctionConfigStore { - return reconciler.NewFunctionConfigStore("", "") +func newTestFunctionConfigStore() *functionconfigs.FunctionConfigStore { + return functionconfigs.NewFunctionConfigStore("", "") } diff --git a/pkg/engine/options.go b/pkg/engine/options.go index 31d827df9..536c0978f 100644 --- a/pkg/engine/options.go +++ b/pkg/engine/options.go @@ -19,7 +19,7 @@ import ( "github.com/kptdev/kpt/pkg/fn" "github.com/kptdev/kpt/pkg/lib/runneroptions" - "github.com/kptdev/porch/controllers/functionconfigs/reconciler" + "github.com/kptdev/porch/controllers/functionconfigs" cachetypes "github.com/kptdev/porch/pkg/cache/types" "github.com/kptdev/porch/pkg/repository" ) @@ -44,7 +44,7 @@ func WithCache(cache cachetypes.Cache) EngineOption { }) } -func WithBuiltinFunctionRuntime(functionConfigStore *reconciler.FunctionConfigStore) EngineOption { +func WithBuiltinFunctionRuntime(functionConfigStore *functionconfigs.FunctionConfigStore) EngineOption { return EngineOptionFunc(func(engine *cadEngine) error { runtime := newBuiltinRuntime(functionConfigStore) if engine.taskHandler.GetRuntime() == nil { diff --git a/pkg/registry/porch/packagerevision_test.go b/pkg/registry/porch/packagerevision_test.go index 2ae141505..da60bf2c4 100644 --- a/pkg/registry/porch/packagerevision_test.go +++ b/pkg/registry/porch/packagerevision_test.go @@ -118,8 +118,15 @@ info: func setup(t *testing.T) (mockClient *mockclient.MockClient, mockEngine *mockengine.MockCaDEngine) { mockClient = mockclient.NewMockClient(t) - packagerevisions.coreClient = mockClient mockEngine = mockengine.NewMockCaDEngine(t) + // Watches start a goroutine against the shared packagerevisions receiver. + // Register ObjectCache before publishing the mock so a late watch cannot + // observe an unstubbed engine. + mockWatcherManager := mockengine.NewMockWatcherManager(t) + mockEngine.On("ObjectCache").Return(mockWatcherManager).Maybe() + mockWatcherManager.On("WatchPackageRevisions", mock.Anything, mock.Anything, mock.Anything). + Return(fmt.Errorf("watch not configured for this test")).Maybe() + packagerevisions.coreClient = mockClient packagerevisions.cad = mockEngine mockClient.On("List", mock.Anything, mock.Anything, mock.Anything).Return(func(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { list.(*configapi.RepositoryList).Items = []configapi.Repository{ @@ -472,8 +479,12 @@ func TestWatch(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _, err := packagerevisions.Watch(ctx, &internalversion.ListOptions{}) + w, err := packagerevisions.Watch(ctx, &internalversion.ListOptions{}) require.NoError(t, err) + w.Stop() + for range w.ResultChan() { + // Drain until the watch goroutine exits and closes the channel. + } //========================================================================================= diff --git a/test/e2e/api/advanced_test.go b/test/e2e/api/advanced_test.go index bdcbeb899..cfad91a68 100644 --- a/test/e2e/api/advanced_test.go +++ b/test/e2e/api/advanced_test.go @@ -22,7 +22,7 @@ import ( porchapi "github.com/kptdev/porch/api/porch/v1alpha1" "github.com/kptdev/porch/pkg/repository" - suiteutils "github.com/kptdev/porch/test/e2e/suiteutils" + "github.com/kptdev/porch/test/e2e/suiteutils" "github.com/stretchr/testify/assert" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" @@ -558,16 +558,7 @@ func (t *PorchSuite) TestPackageMetadataFromKptfile() { {ConditionType: "Ready"}, {ConditionType: "Deployed"}, } - // Note: Gates order may vary due to Kptfile serialization/deserialization - // Compare as sets by checking presence and count - t.Require().Len(clonePr.Spec.ReadinessGates, len(expectedGates), "should have same number of gates") - actualGateTypes := make(map[string]bool) - for _, g := range clonePr.Spec.ReadinessGates { - actualGateTypes[g.ConditionType] = true - } - for _, g := range expectedGates { - t.Require().True(actualGateTypes[g.ConditionType], "expected gate %s not found", g.ConditionType) - } + t.Require().ElementsMatch(expectedGates, clonePr.Spec.ReadinessGates) var packageResources porchapi.PackageRevisionResources t.GetF(client.ObjectKeyFromObject(clonePr), &packageResources) @@ -618,17 +609,7 @@ func (t *PorchSuite) TestPackageMetadataFromKptfile() { t.Require().True(ok, "annotation key %s should exist", k) t.Require().Equal(v, actual, "annotation %s value should match", k) } - - // Note: Gates order may vary due to Kptfile serialization/deserialization - // Compare as sets by checking presence and count - t.Require().Len(mainPr.Spec.ReadinessGates, len(expectedGates), "should have same number of gates") - actualGateTypes := make(map[string]bool) - for _, g := range mainPr.Spec.ReadinessGates { - actualGateTypes[g.ConditionType] = true - } - for _, g := range expectedGates { - t.Require().True(actualGateTypes[g.ConditionType], "expected gate %s not found", g.ConditionType) - } + t.Require().ElementsMatch(expectedGates, mainPr.Spec.ReadinessGates, "main revision ReadinessGates should match v1") }) }