Implement Prioriy and Serial settings for discovered handlers

Signed-off-by: Danil-Grigorev <[email protected]>

Author: Danil-Grigorev

config/crd/bases/runtime.cluster.x-k8s.io_extensionconfigs.yaml

@@ -131,6 +131,14 @@ type ExtensionHandler struct {
 	// Defaults to Fail if not set.
 	// +optional
 	FailurePolicy *FailurePolicy `json:"failurePolicy,omitempty"`
+
+	// Priority defines the order in which this handler will be invoked. Hooks are executed in the descending order.
+	// +optional
+	Priority int32 `json:"priority,omitempty"`
+
+	// Serial defines if the hook is allowed to run in parallel with others.
+	// +optional
+	Serial bool `json:"serial,omitempty"`
 }
 
 // GroupVersionHook defines the runtime hook when the ExtensionHandler is called.

exp/runtime/api/v1alpha1/extensionconfig_types.go

@@ -63,6 +63,14 @@ type ExtensionHandler struct {
 	// failurePolicy defines how failures in calls to the ExtensionHandler should be handled by a client.
 	// This is defaulted to FailurePolicyFail if not defined.
 	FailurePolicy *FailurePolicy `json:"failurePolicy,omitempty"`
+
+	// Priority defines the order in which this handler will be invoked. Hooks are executed in the descending order.
+	// +optional
+	Priority int32 `json:"priority,omitempty"`
+
+	// Serial defines if the hook is allowed to run in parallel with others.
+	// +optional
+	Serial bool `json:"serial,omitempty"`
 }
 
 // GroupVersionHook defines the runtime hook when the ExtensionHandler is called.

exp/runtime/hooks/api/v1alpha1/discovery_types.go

@@ -229,6 +229,10 @@ func TestExtensionReconciler_discoverExtensionConfig(t *testing.T) {
 		handlers := discoveredExtensionConfig.Status.Handlers
 		g.Expect(handlers).To(HaveLen(1))
 		g.Expect(handlers[0].Name).To(Equal("first.ext1"))
+		g.Expect(handlers[0].RequestHook.Hook).To(Equal("FakeHook"))
+		g.Expect(handlers[0].RequestHook.APIVersion).To(Equal("test.runtime.cluster.x-k8s.io/v1alpha1"))
+		g.Expect(handlers[0].Serial).To(BeTrue())
+		g.Expect(handlers[0].Priority).To(Equal(int32(100)))
 
 		// Expect exactly one condition and expect the condition to have type RuntimeExtensionDiscoveredCondition and
 		// Status true.
@@ -345,6 +349,8 @@ func discoveryHandler(handlerList ...string) func(http.ResponseWriter, *http.Req
 				Hook:       "FakeHook",
 				APIVersion: fakev1alpha1.GroupVersion.String(),
 			},
+			Serial:   true,
+			Priority: 100,
 		})
 	}
 	response := &runtimehooksv1.DiscoveryResponse{

exp/runtime/hooks/api/v1alpha1/zz_generated.openapi.go

@@ -143,6 +143,8 @@ func (c *client) Discover(ctx context.Context, extensionConfig *runtimev1.Extens
 				},
 				TimeoutSeconds: handler.TimeoutSeconds,
 				FailurePolicy:  (*runtimev1.FailurePolicy)(handler.FailurePolicy),
+				Priority:       handler.Priority,
+				Serial:         handler.Serial,
 			},
 		)
 	}
@@ -193,6 +195,7 @@ func (c *client) CallAllExtensions(ctx context.Context, hook runtimecatalog.Hook
 
 	log.V(4).Info(fmt.Sprintf("Calling all extensions of hook %q", hookName))
 	responses := []runtimehooksv1.ResponseObject{}
+	retry := false
 	for _, registration := range registrations {
 		// Creates a new instance of the response parameter.
 		responseObject, err := c.catalog.NewResponse(gvh)
@@ -212,13 +215,28 @@ func (c *client) CallAllExtensions(ctx context.Context, hook runtimecatalog.Hook
 			continue
 		}
 
+		if registration.Serial && retry {
+			log.V(5).Info(fmt.Sprintf("Serial handler %q waits for blocking response to complete", registration.Name))
+			break
+		}
+
 		err = c.CallExtension(ctx, hook, forObject, registration.Name, request, tmpResponse)
 		// If one of the extension handlers fails lets short-circuit here and return early.
 		if err != nil {
 			log.Error(err, "failed to call extension handlers")
 			return errors.Wrapf(err, "failed to call extension handlers for hook %q", gvh.GroupHook())
 		}
+
+		if retryResponse, isRetry := tmpResponse.(runtimehooksv1.RetryResponseObject); isRetry && !retry && retryResponse.GetRetryAfterSeconds() > 0 {
+			retry = isRetry
+		}
+
 		responses = append(responses, tmpResponse)
+
+		if registration.Serial && retry {
+			log.V(5).Info(fmt.Sprintf("Serial handler %q is blocking hook until it is completed", registration.Name))
+			break
+		}
 	}
 
 	// Aggregate all responses into a single response.

exp/runtime/internal/controllers/extensionconfig_controller_test.go

@@ -984,6 +984,17 @@ func TestClient_CallAllExtensions(t *testing.T) {
 		},
 	}
 
+	secondBlockingConfig := extensionConfig.DeepCopy()
+	secondBlockingConfig.Status.Handlers[0].Priority = 2
+	secondBlockingConfig.Status.Handlers[1].Priority = 1
+	secondBlockingConfig.Status.Handlers[1].Serial = true
+	secondBlockingConfig.Status.Handlers[0].RequestHook.Hook = "RetryableFakeHook"
+	secondBlockingConfig.Status.Handlers[1].RequestHook.Hook = "RetryableFakeHook"
+	secondBlockingConfig.Status.Handlers[2].RequestHook.Hook = "RetryableFakeHook"
+
+	secondBlockingWithPriorityConfig := secondBlockingConfig.DeepCopy()
+	secondBlockingWithPriorityConfig.Status.Handlers[1].Priority = 3
+
 	type args struct {
 		hook     runtimecatalog.Hook
 		request  runtimehooksv1.RequestObject
@@ -1072,6 +1083,55 @@ func TestClient_CallAllExtensions(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name:                       "should succeed and wait on previous blocking responses for serial handler",
+			registeredExtensionConfigs: []runtimev1.ExtensionConfig{*secondBlockingConfig},
+			testServer: testServerConfig{
+				start: true,
+				responses: map[string]testServerResponse{
+					"/test.runtime.cluster.x-k8s.io/v1alpha1/retryablefakehook/first-extension.*": retryResponse(runtimehooksv1.ResponseStatusSuccess, 1),
+					// second and third extension has no handler.
+				},
+			},
+			args: args{
+				hook:     fakev1alpha1.RetryableFakeHook,
+				request:  &fakev1alpha1.RetryableFakeRequest{},
+				response: &fakev1alpha1.RetryableFakeResponse{},
+			},
+		},
+		{
+			name:                       "should succeed and wait on blocking serial handler",
+			registeredExtensionConfigs: []runtimev1.ExtensionConfig{*secondBlockingConfig},
+			testServer: testServerConfig{
+				start: true,
+				responses: map[string]testServerResponse{
+					"/test.runtime.cluster.x-k8s.io/v1alpha1/retryablefakehook/first-extension.*":  response(runtimehooksv1.ResponseStatusSuccess),
+					"/test.runtime.cluster.x-k8s.io/v1alpha1/retryablefakehook/second-extension.*": retryResponse(runtimehooksv1.ResponseStatusSuccess, 1),
+					// third-extension has no handler.
+				},
+			},
+			args: args{
+				hook:     fakev1alpha1.RetryableFakeHook,
+				request:  &fakev1alpha1.RetryableFakeRequest{},
+				response: &fakev1alpha1.RetryableFakeResponse{},
+			},
+		},
+		{
+			name:                       "should succeed and wait on blocking serial handler, which is called with priority",
+			registeredExtensionConfigs: []runtimev1.ExtensionConfig{*secondBlockingWithPriorityConfig},
+			testServer: testServerConfig{
+				start: true,
+				responses: map[string]testServerResponse{
+					"/test.runtime.cluster.x-k8s.io/v1alpha1/retryablefakehook/second-extension.*": retryResponse(runtimehooksv1.ResponseStatusSuccess, 1),
+					// second and third extension has no handler.
+				},
+			},
+			args: args{
+				hook:     fakev1alpha1.RetryableFakeHook,
+				request:  &fakev1alpha1.RetryableFakeRequest{},
+				response: &fakev1alpha1.RetryableFakeResponse{},
+			},
+		},
 		{
 			name:                       "should fail when one of the ExtensionHandlers returns 404",
 			registeredExtensionConfigs: []runtimev1.ExtensionConfig{extensionConfig},
@@ -1317,6 +1377,20 @@ func response(status runtimehooksv1.ResponseStatus) testServerResponse {
 	}
 }
 
+func retryResponse(status runtimehooksv1.ResponseStatus, retrySeconds int32) testServerResponse {
+	return testServerResponse{
+		response: &fakev1alpha1.RetryableFakeResponse{
+			CommonRetryResponse: runtimehooksv1.CommonRetryResponse{
+				RetryAfterSeconds: retrySeconds,
+				CommonResponse: runtimehooksv1.CommonResponse{
+					Status: status,
+				},
+			},
+		},
+		responseStatusCode: http.StatusOK,
+	}
+}
+
 func createSecureTestServer(server testServerConfig, callbacks ...func()) *httptest.Server {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
@@ -1335,7 +1409,10 @@ func createSecureTestServer(server testServerConfig, callbacks ...func()) *httpt
 				panic(err)
 			}
 			w.WriteHeader(resp.responseStatusCode)
-			_, _ = w.Write(respBody)
+			_, err = w.Write(respBody)
+			if err != nil {
+				panic(err)
+			}
 			return
 		}
 

internal/runtime/client/client.go

@@ -17,6 +17,7 @@ limitations under the License.
 package registry
 
 import (
+	"sort"
 	"sync"
 
 	"github.com/pkg/errors"
@@ -84,6 +85,13 @@ type ExtensionRegistration struct {
 
 	// Settings captures additional information sent in call to the RuntimeExtensions.
 	Settings map[string]string
+
+	// Priority defines the order in which this handler will be invoked. Hooks are executed in the descending order.
+	Priority int32
+
+	// Serial defines if the hook should be executed serially. This ensures that previously pending hooks are finished
+	// first, as well as current hook reached completion, before moving to the next one.
+	Serial bool
 }
 
 // extensionRegistry is an implementation of ExtensionRegistry.
@@ -210,6 +218,11 @@ func (r *extensionRegistry) List(gh runtimecatalog.GroupHook) ([]*ExtensionRegis
 			l = append(l, registration)
 		}
 	}
+
+	sort.SliceStable(l, func(i, j int) bool {
+		return l[i].Priority > l[j].Priority
+	})
+
 	return l, nil
 }
 
@@ -263,6 +276,8 @@ func (r *extensionRegistry) add(extensionConfig *runtimev1.ExtensionConfig) erro
 			TimeoutSeconds:    e.TimeoutSeconds,
 			FailurePolicy:     e.FailurePolicy,
 			Settings:          extensionConfig.Spec.Settings,
+			Priority:          e.Priority,
+			Serial:            e.Serial,
 		})
 	}