Skip to content
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 2 additions & 12 deletions cmd/epp/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,16 +30,12 @@ import (
"go.uber.org/zap/zapcore"
"google.golang.org/grpc"
healthPb "google.golang.org/grpc/health/grpc_health_v1"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
"k8s.io/component-base/metrics/legacyregistry"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/log/zap"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend"
"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/vllm"
Expand Down Expand Up @@ -97,15 +93,9 @@ var (
"are assumed to be named tls.crt and tls.key, respectively. If not set, and secureServing is enabled, "+
"then a self-signed certificate is used.")

scheme = runtime.NewScheme()
setupLog = ctrl.Log.WithName("setup")
)

func init() {
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
utilruntime.Must(v1alpha2.AddToScheme(scheme))
}

func main() {
if err := run(); err != nil {
os.Exit(1)
Expand Down Expand Up @@ -140,9 +130,9 @@ func run() error {
return err
}

mgr, err := ctrl.NewManager(cfg, ctrl.Options{Scheme: scheme})
mgr, err := runserver.NewDefaultManager(*poolNamespace, *poolName)
if err != nil {
setupLog.Error(err, "Failed to create controller manager", "config", cfg)
setupLog.Error(err, "Failed to create controller manager")
return err
}

Expand Down
4 changes: 1 addition & 3 deletions pkg/epp/controller/inferencemodel_reconciler.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"fmt"

"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
ctrl "sigs.k8s.io/controller-runtime"
Expand All @@ -36,7 +35,6 @@ import (

type InferenceModelReconciler struct {
client.Client
Scheme *runtime.Scheme
Record record.EventRecorder
Datastore datastore.Datastore
PoolNamespacedName types.NamespacedName
Expand Down Expand Up @@ -128,5 +126,5 @@ func (c *InferenceModelReconciler) SetupWithManager(ctx context.Context, mgr ctr
}

func (c *InferenceModelReconciler) eventPredicate(infModel *v1alpha2.InferenceModel) bool {
return (infModel.Spec.PoolRef.Name == v1alpha2.ObjectName(c.PoolNamespacedName.Name)) && (infModel.GetNamespace() == c.PoolNamespacedName.Namespace)
return string(infModel.Spec.PoolRef.Name) == c.PoolNamespacedName.Name
}
1 change: 0 additions & 1 deletion pkg/epp/controller/inferencemodel_reconciler_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,6 @@ func TestInferenceModelReconciler(t *testing.T) {
datastore := datastore.NewFakeDatastore(nil, test.modelsInStore, pool)
reconciler := &InferenceModelReconciler{
Client: fakeClient,
Scheme: scheme,
Record: record.NewFakeRecorder(10),
Datastore: datastore,
PoolNamespacedName: types.NamespacedName{Name: pool.Name, Namespace: pool.Namespace},
Expand Down
6 changes: 0 additions & 6 deletions pkg/epp/controller/inferencepool_reconciler.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,11 @@ import (
"reflect"

"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
Expand All @@ -38,7 +36,6 @@ import (
// will have the proper controller that will create/manage objects on behalf of the server pool.
type InferencePoolReconciler struct {
client.Client
Scheme *runtime.Scheme
Record record.EventRecorder
PoolNamespacedName types.NamespacedName
Datastore datastore.Datastore
Expand Down Expand Up @@ -90,8 +87,5 @@ func (c *InferencePoolReconciler) updateDatastore(ctx context.Context, newPool *
func (c *InferencePoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&v1alpha2.InferencePool{}).
WithEventFilter(predicate.NewPredicateFuncs(func(object client.Object) bool {
return (object.GetNamespace() == c.PoolNamespacedName.Namespace) && (object.GetName() == c.PoolNamespacedName.Name)
})).
Complete(c)
}
9 changes: 2 additions & 7 deletions pkg/epp/controller/pod_reconciler.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import (
"github.com/go-logr/logr"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
ctrl "sigs.k8s.io/controller-runtime"
Expand All @@ -35,19 +34,15 @@ import (
type PodReconciler struct {
client.Client
Datastore datastore.Datastore
Scheme *runtime.Scheme
Record record.EventRecorder
}

func (c *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
logger := log.FromContext(ctx)
inferencePool, err := c.Datastore.PoolGet()
if err != nil {
logger.V(logutil.TRACE).Info("Skipping reconciling Pod because the InferencePool is not available yet", "error", err)
if !c.Datastore.PoolHasSynced() {
logger.V(logutil.TRACE).Info("Skipping reconciling Pod because the InferencePool is not available yet")
// When the inferencePool is initialized it lists the appropriate pods and populates the datastore, so no need to requeue.
return ctrl.Result{}, nil
} else if inferencePool.Namespace != req.Namespace {
return ctrl.Result{}, nil
}

logger.V(logutil.VERBOSE).Info("Pod being reconciled", "name", req.NamespacedName)
Expand Down
78 changes: 78 additions & 0 deletions pkg/epp/server/controller_manager.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
/*
Copyright 2025 The Kubernetes 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 server

import (
"fmt"

corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/cache"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha2"
)

var scheme = runtime.NewScheme()

func init() {
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
utilruntime.Must(v1alpha2.AddToScheme(scheme))
}

// NewDefaultManager creates a new controller manager with default configuration.
func NewDefaultManager(namespace, name string) (ctrl.Manager, error) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let me see if there is a trap

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated, we should add more test cases later

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

perhaps you forgot to push the update?

// Init runtime.
cfg, err := ctrl.GetConfig()
if err != nil {
return nil, fmt.Errorf("failed to create rest config: %v", err)
}

manager, err := ctrl.NewManager(cfg, ctrl.Options{
Scheme: scheme,
Cache: cache.Options{
ByObject: map[client.Object]cache.ByObject{
&corev1.Pod{}: {
Namespaces: map[string]cache.Config{
namespace: {},
},
},
&v1alpha2.InferencePool{}: {
Namespaces: map[string]cache.Config{
namespace: {
FieldSelector: fields.SelectorFromSet(fields.Set{
"metadata.name": name,
}),
},
},
},
&v1alpha2.InferenceModel{}: {
Namespaces: map[string]cache.Config{
namespace: {},
},
},
},
},
})
if err != nil {
return nil, fmt.Errorf("failed to create controller manager: %v", err)
}
return manager, nil
}
3 changes: 0 additions & 3 deletions pkg/epp/server/runserver.go
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,6 @@ func (r *ExtProcServerRunner) SetupWithManager(ctx context.Context, mgr ctrl.Man
// Create the controllers and register them with the manager
if err := (&controller.InferencePoolReconciler{
Datastore: r.Datastore,
Scheme: mgr.GetScheme(),
Client: mgr.GetClient(),
PoolNamespacedName: types.NamespacedName{
Name: r.PoolName,
Expand All @@ -102,7 +101,6 @@ func (r *ExtProcServerRunner) SetupWithManager(ctx context.Context, mgr ctrl.Man

if err := (&controller.InferenceModelReconciler{
Datastore: r.Datastore,
Scheme: mgr.GetScheme(),
Client: mgr.GetClient(),
PoolNamespacedName: types.NamespacedName{
Name: r.PoolName,
Expand All @@ -115,7 +113,6 @@ func (r *ExtProcServerRunner) SetupWithManager(ctx context.Context, mgr ctrl.Man

if err := (&controller.PodReconciler{
Datastore: r.Datastore,
Scheme: mgr.GetScheme(),
Client: mgr.GetClient(),
Record: mgr.GetEventRecorderFor("pod"),
}).SetupWithManager(mgr); err != nil {
Expand Down