Merge pull request kubernetes-sigs#8945 from sbueringer/pr-improve-locking

🌱 test/e2e/in-memory: improve locking, return errors instead of panic

Author: k8s-ci-robot

test/infrastructure/inmemory/internal/server/api/handler.go

@@ -542,7 +542,7 @@ func requestToGVK(req *restful.Request) (*schema.GroupVersionKind, error) {
 	}
 	gv, err := schema.ParseGroupVersion(resourceList.GroupVersion)
 	if err != nil {
-		panic(fmt.Sprintf("invalid group version in APIResourceList: %s", resourceList.GroupVersion))
+		return nil, errors.Errorf("invalid group version in APIResourceList: %s", resourceList.GroupVersion)
 	}
 
 	resource := req.PathParameter("resource")

test/infrastructure/inmemory/internal/server/api/watch.go

@@ -97,7 +97,8 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful.
 	h.log.Info(fmt.Sprintf("Serving Watch for %v", req.Request.URL))
 	// With an unbuffered event channel RemoveEventHandler could be blocked because it requires a lock on the informer.
 	// When Run stops reading from the channel the informer could be blocked with an unbuffered chanel and then RemoveEventHandler never goes through.
-	events := make(chan *Event, 10)
+	// 1000 is used to avoid deadlocks in clusters with a higher number of Machines/Nodes.
+	events := make(chan *Event, 1000)
 	watcher := &WatchEventDispatcher{
 		resourceGroup: resourceGroup,
 		events:        events,
@@ -109,8 +110,8 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful.
 
 	// Defer cleanup which removes the event handler and ensures the channel is empty of events.
 	defer func() {
-		reterr = i.RemoveEventHandler(watcher)
 		// Doing this to ensure the channel is empty.
+		// This reduces the probability of a deadlock when removing the event handler.
 	L:
 		for {
 			select {
@@ -119,6 +120,8 @@ func (h *apiServerHandler) watchForResource(req *restful.Request, resp *restful.
 				break L
 			}
 		}
+		reterr = i.RemoveEventHandler(watcher)
+		// Note: After we removed the handler, no new events will be written to the events channel.
 	}()
 
 	return watcher.Run(ctx, queryTimeout, resp)

test/infrastructure/inmemory/internal/server/etcd/handler.go

@@ -98,27 +98,27 @@ func (m *maintenanceServer) Status(ctx context.Context, _ *pb.StatusRequest) (*p
 }
 
 func (m *maintenanceServer) Defragment(_ context.Context, _ *pb.DefragmentRequest) (*pb.DefragmentResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: Defragment")
 }
 
 func (m *maintenanceServer) Hash(_ context.Context, _ *pb.HashRequest) (*pb.HashResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: Hash")
 }
 
 func (m *maintenanceServer) HashKV(_ context.Context, _ *pb.HashKVRequest) (*pb.HashKVResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: HashKV")
 }
 
 func (m *maintenanceServer) Snapshot(_ *pb.SnapshotRequest, _ pb.Maintenance_SnapshotServer) error {
-	panic("implement me")
+	return fmt.Errorf("not implemented: Snapshot")
 }
 
 func (m *maintenanceServer) MoveLeader(_ context.Context, _ *pb.MoveLeaderRequest) (*pb.MoveLeaderResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: MoveLeader")
 }
 
 func (m *maintenanceServer) Downgrade(_ context.Context, _ *pb.DowngradeRequest) (*pb.DowngradeResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: Downgrade")
 }
 
 // clusterServerServer implements the ClusterServer grpc server.
@@ -127,15 +127,15 @@ type clusterServerServer struct {
 }
 
 func (c *clusterServerServer) MemberAdd(_ context.Context, _ *pb.MemberAddRequest) (*pb.MemberAddResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: MemberAdd")
 }
 
 func (c *clusterServerServer) MemberRemove(_ context.Context, _ *pb.MemberRemoveRequest) (*pb.MemberRemoveResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: MemberRemove")
 }
 
 func (c *clusterServerServer) MemberUpdate(_ context.Context, _ *pb.MemberUpdateRequest) (*pb.MemberUpdateResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: MemberUpdate")
 }
 
 func (c *clusterServerServer) MemberList(ctx context.Context, _ *pb.MemberListRequest) (*pb.MemberListResponse, error) {
@@ -155,7 +155,7 @@ func (c *clusterServerServer) MemberList(ctx context.Context, _ *pb.MemberListRe
 }
 
 func (c *clusterServerServer) MemberPromote(_ context.Context, _ *pb.MemberPromoteRequest) (*pb.MemberPromoteResponse, error) {
-	panic("implement me")
+	return nil, fmt.Errorf("not implemented: MemberPromote")
 }
 
 type baseServer struct {