Renamed the post response plugins to not include the word post.

Author: BenjaminBraunDev

pkg/epp/requestcontrol/director.go

@@ -59,29 +59,37 @@ type SaturationDetector interface {
 	IsSaturated(ctx context.Context, candidatePods []backendmetrics.PodMetrics) bool
 }
 
+type RequestControlPlugins struct {
+	preRequestPlugins        []PreRequest
+	responseReceivedPlugins  []ResponseReceived
+	responseStreamingPlugins []ResponseStreaming
+	responseCompletePlugins  []ResponseComplete
+}
+
 // NewDirectorWithConfig creates a new Director instance with all dependencies.
 func NewDirectorWithConfig(datastore Datastore, scheduler Scheduler, saturationDetector SaturationDetector, config *Config) *Director {
+	RCPlugins := RequestControlPlugins{
+		preRequestPlugins:        config.preRequestPlugins,
+		responseReceivedPlugins:  config.responseReceivedPlugins,
+		responseStreamingPlugins: config.responseStreamingPlugins,
+		responseCompletePlugins:  config.responseCompletePlugins,
+	}
+
 	return &Director{
-		datastore:                    datastore,
-		scheduler:                    scheduler,
-		saturationDetector:           saturationDetector,
-		preRequestPlugins:            config.preRequestPlugins,
-		postResponseReceivedPlugins:  config.postResponseReceivedPlugins,
-		postResponseStreamingPlugins: config.postResponseStreamingPlugins,
-		postResponseCompletePlugins:  config.postResponseCompletePlugins,
-		defaultPriority:              0, // define default priority explicitly
+		datastore:             datastore,
+		scheduler:             scheduler,
+		saturationDetector:    saturationDetector,
+		requestControlPlugins: RCPlugins,
+		defaultPriority:       0, // define default priority explicitly
 	}
 }
 
 // Director orchestrates the request handling flow, including scheduling.
 type Director struct {
-	datastore                    Datastore
-	scheduler                    Scheduler
-	saturationDetector           SaturationDetector
-	preRequestPlugins            []PreRequest
-	postResponseReceivedPlugins  []PostResponseReceived
-	postResponseStreamingPlugins []PostResponseStreaming
-	postResponseCompletePlugins  []PostResponseComplete
+	datastore             Datastore
+	scheduler             Scheduler
+	saturationDetector    SaturationDetector
+	requestControlPlugins RequestControlPlugins
 	// we just need a pointer to an int variable since priority is a pointer in InferenceObjective
 	// no need to set this in the constructor, since the value we want is the default int val
 	// and value types cannot be nil
@@ -291,7 +299,7 @@ func (d *Director) HandleResponseReceived(ctx context.Context, reqCtx *handlers.
 
 	// TODO: to extend fallback functionality, handle cases where target pod is unavailable
 	// https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/1224
-	d.runPostResponseReceivedPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
+	d.runResponseReceivedPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
 
 	return reqCtx, nil
 }
@@ -305,7 +313,7 @@ func (d *Director) HandleResponseBodyStreaming(ctx context.Context, reqCtx *hand
 		Headers:   reqCtx.Response.Headers,
 	}
 
-	d.runPostResponseStreamingPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
+	d.runResponseStreamingPlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
 	logger.V(logutil.TRACE).Info("Exiting HandleResponseBodyChunk")
 	return reqCtx, nil
 }
@@ -319,7 +327,7 @@ func (d *Director) HandleResponseBodyComplete(ctx context.Context, reqCtx *handl
 		Headers:   reqCtx.Response.Headers,
 	}
 
-	d.runPostResponseCompletePlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
+	d.runResponseCompletePlugins(ctx, reqCtx.SchedulingRequest, response, reqCtx.TargetPod)
 
 	logger.V(logutil.DEBUG).Info("Exiting HandleResponseBodyComplete")
 	return reqCtx, nil
@@ -338,7 +346,7 @@ func (d *Director) GetRandomPod() *backend.Pod {
 func (d *Director) runPreRequestPlugins(ctx context.Context, request *schedulingtypes.LLMRequest,
 	schedulingResult *schedulingtypes.SchedulingResult, targetPort int) {
 	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
-	for _, plugin := range d.preRequestPlugins {
+	for _, plugin := range d.requestControlPlugins.preRequestPlugins {
 		loggerDebug.Info("Running pre-request plugin", "plugin", plugin.TypedName())
 		before := time.Now()
 		plugin.PreRequest(ctx, request, schedulingResult, targetPort)
@@ -347,34 +355,34 @@ func (d *Director) runPreRequestPlugins(ctx context.Context, request *scheduling
 	}
 }
 
-func (d *Director) runPostResponseReceivedPlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (d *Director) runResponseReceivedPlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
-	for _, plugin := range d.postResponseReceivedPlugins {
+	for _, plugin := range d.requestControlPlugins.responseReceivedPlugins {
 		loggerDebug.Info("Running post-response plugin", "plugin", plugin.TypedName())
 		before := time.Now()
-		plugin.PostResponseReceived(ctx, request, response, targetPod)
-		metrics.RecordPluginProcessingLatency(PostResponseReceivedExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
+		plugin.ResponseReceived(ctx, request, response, targetPod)
+		metrics.RecordPluginProcessingLatency(ResponseReceivedExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
 		loggerDebug.Info("Completed running post-response plugin successfully", "plugin", plugin.TypedName())
 	}
 }
 
-func (d *Director) runPostResponseStreamingPlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (d *Director) runResponseStreamingPlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	loggerTrace := log.FromContext(ctx).V(logutil.TRACE)
-	for _, plugin := range d.postResponseStreamingPlugins {
+	for _, plugin := range d.requestControlPlugins.responseStreamingPlugins {
 		loggerTrace.Info("Running post-response chunk plugin", "plugin", plugin.TypedName())
 		before := time.Now()
-		plugin.PostResponseStreaming(ctx, request, response, targetPod)
-		metrics.RecordPluginProcessingLatency(PostResponseStreamingExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
+		plugin.ResponseStreaming(ctx, request, response, targetPod)
+		metrics.RecordPluginProcessingLatency(ResponseStreamingExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
 	}
 }
 
-func (d *Director) runPostResponseCompletePlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (d *Director) runResponseCompletePlugins(ctx context.Context, request *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	loggerDebug := log.FromContext(ctx).V(logutil.DEBUG)
-	for _, plugin := range d.postResponseCompletePlugins {
+	for _, plugin := range d.requestControlPlugins.responseCompletePlugins {
 		loggerDebug.Info("Running post-response complete plugin", "plugin", plugin.TypedName())
 		before := time.Now()
-		plugin.PostResponseComplete(ctx, request, response, targetPod)
-		metrics.RecordPluginProcessingLatency(PostResponseCompleteExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
+		plugin.ResponseComplete(ctx, request, response, targetPod)
+		metrics.RecordPluginProcessingLatency(ResponseCompleteExtensionPoint, plugin.TypedName().Type, plugin.TypedName().Name, time.Since(before))
 		loggerDebug.Info("Completed running post-response complete plugin successfully", "plugin", plugin.TypedName())
 	}
 }

pkg/epp/requestcontrol/director_test.go

@@ -593,12 +593,12 @@ func TestGetRandomPod(t *testing.T) {
 }
 
 func TestDirector_HandleResponseReceived(t *testing.T) {
-	pr1 := newTestPostResponseReceived("pr1")
+	pr1 := newTestResponseReceived("pr1")
 
 	ctx := logutil.NewTestLoggerIntoContext(context.Background())
 	ds := datastore.NewDatastore(t.Context(), nil)
 	mockSched := &mockScheduler{}
-	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithPostResponseReceivedPlugins(pr1))
+	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithResponseReceivedPlugins(pr1))
 
 	reqCtx := &handlers.RequestContext{
 		Request: &handlers.Request{
@@ -630,12 +630,12 @@ func TestDirector_HandleResponseReceived(t *testing.T) {
 }
 
 func TestDirector_HandleResponseStreaming(t *testing.T) {
-	ps1 := newTestPostResponseStreaming("ps1")
+	ps1 := newTestResponseStreaming("ps1")
 
 	ctx := logutil.NewTestLoggerIntoContext(context.Background())
 	ds := datastore.NewDatastore(t.Context(), nil)
 	mockSched := &mockScheduler{}
-	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithPostResponseStreamingPlugins(ps1))
+	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithResponseStreamingPlugins(ps1))
 
 	reqCtx := &handlers.RequestContext{
 		Request: &handlers.Request{
@@ -666,12 +666,12 @@ func TestDirector_HandleResponseStreaming(t *testing.T) {
 }
 
 func TestDirector_HandleResponseComplete(t *testing.T) {
-	pc1 := newTestPostResponseComplete("pc1")
+	pc1 := newTestResponseComplete("pc1")
 
 	ctx := logutil.NewTestLoggerIntoContext(context.Background())
 	ds := datastore.NewDatastore(t.Context(), nil)
 	mockSched := &mockScheduler{}
-	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithPostResponseCompletePlugins(pc1))
+	director := NewDirectorWithConfig(ds, mockSched, nil, NewConfig().WithResponseCompletePlugins(pc1))
 
 	reqCtx := &handlers.RequestContext{
 		Request: &handlers.Request{
@@ -702,70 +702,70 @@ func TestDirector_HandleResponseComplete(t *testing.T) {
 }
 
 const (
-	testPostResponseReceivedType = "test-post-response"
-	testPostStreamingType        = "test-post-streaming"
-	testPostCompleteType         = "test-post-complete"
+	testResponseReceivedType = "test-post-response"
+	testPostStreamingType    = "test-post-streaming"
+	testPostCompleteType     = "test-post-complete"
 )
 
-type testPostResponseReceived struct {
+type testResponseReceived struct {
 	tn                      plugins.TypedName
 	lastRespOnResponse      *Response
 	lastTargetPodOnResponse string
 }
 
-type testPostResponseStreaming struct {
+type testResponseStreaming struct {
 	tn                       plugins.TypedName
 	lastRespOnStreaming      *Response
 	lastTargetPodOnStreaming string
 }
 
-type testPostResponseComplete struct {
+type testResponseComplete struct {
 	tn                      plugins.TypedName
 	lastRespOnComplete      *Response
 	lastTargetPodOnComplete string
 }
 
-func newTestPostResponseReceived(name string) *testPostResponseReceived {
-	return &testPostResponseReceived{
-		tn: plugins.TypedName{Type: testPostResponseReceivedType, Name: name},
+func newTestResponseReceived(name string) *testResponseReceived {
+	return &testResponseReceived{
+		tn: plugins.TypedName{Type: testResponseReceivedType, Name: name},
 	}
 }
 
-func newTestPostResponseStreaming(name string) *testPostResponseStreaming {
-	return &testPostResponseStreaming{
+func newTestResponseStreaming(name string) *testResponseStreaming {
+	return &testResponseStreaming{
 		tn: plugins.TypedName{Type: testPostStreamingType, Name: name},
 	}
 }
 
-func newTestPostResponseComplete(name string) *testPostResponseComplete {
-	return &testPostResponseComplete{
+func newTestResponseComplete(name string) *testResponseComplete {
+	return &testResponseComplete{
 		tn: plugins.TypedName{Type: testPostCompleteType, Name: name},
 	}
 }
 
-func (p *testPostResponseReceived) TypedName() plugins.TypedName {
+func (p *testResponseReceived) TypedName() plugins.TypedName {
 	return p.tn
 }
 
-func (p *testPostResponseStreaming) TypedName() plugins.TypedName {
+func (p *testResponseStreaming) TypedName() plugins.TypedName {
 	return p.tn
 }
 
-func (p *testPostResponseComplete) TypedName() plugins.TypedName {
+func (p *testResponseComplete) TypedName() plugins.TypedName {
 	return p.tn
 }
 
-func (p *testPostResponseReceived) PostResponseReceived(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (p *testResponseReceived) ResponseReceived(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	p.lastRespOnResponse = response
 	p.lastTargetPodOnResponse = targetPod.NamespacedName.String()
 }
 
-func (p *testPostResponseStreaming) PostResponseStreaming(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (p *testResponseStreaming) ResponseStreaming(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	p.lastRespOnStreaming = response
 	p.lastTargetPodOnStreaming = targetPod.NamespacedName.String()
 }
 
-func (p *testPostResponseComplete) PostResponseComplete(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
+func (p *testResponseComplete) ResponseComplete(_ context.Context, _ *schedulingtypes.LLMRequest, response *Response, targetPod *backend.Pod) {
 	p.lastRespOnComplete = response
 	p.lastTargetPodOnComplete = targetPod.NamespacedName.String()
 }

pkg/epp/requestcontrol/plugins.go

@@ -25,10 +25,10 @@ import (
 )
 
 const (
-	PreRequestExtensionPoint            = "PreRequest"
-	PostResponseReceivedExtensionPoint  = "PostResponseReceived"
-	PostResponseStreamingExtensionPoint = "PostResponseStreaming"
-	PostResponseCompleteExtensionPoint  = "PostResponseComplete"
+	PreRequestExtensionPoint        = "PreRequest"
+	ResponseReceivedExtensionPoint  = "ResponseReceived"
+	ResponseStreamingExtensionPoint = "ResponseStreaming"
+	ResponseCompleteExtensionPoint  = "ResponseComplete"
 )
 
 // PreRequest is called by the director after a getting result from scheduling layer and
@@ -38,21 +38,21 @@ type PreRequest interface {
 	PreRequest(ctx context.Context, request *types.LLMRequest, schedulingResult *types.SchedulingResult, targetPort int)
 }
 
-// PostResponseReceived is called by the director after a successful response is sent.
+// ResponseReceived is called by the director after a successful response is sent.
 // The given pod argument is the pod that served the request.
-type PostResponseReceived interface {
+type ResponseReceived interface {
 	plugins.Plugin
-	PostResponseReceived(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
+	ResponseReceived(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
 }
 
-// PostResponseStreaming is called by the director after each chunk of streaming response is sent.
-type PostResponseStreaming interface {
+// ResponseStreaming is called by the director after each chunk of streaming response is sent.
+type ResponseStreaming interface {
 	plugins.Plugin
-	PostResponseStreaming(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
+	ResponseStreaming(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
 }
 
-// PostResponseComplete is called by the director after the complete response is sent.
-type PostResponseComplete interface {
+// ResponseComplete is called by the director after the complete response is sent.
+type ResponseComplete interface {
 	plugins.Plugin
-	PostResponseComplete(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
+	ResponseComplete(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
 }