Skip to content

Bug 1961925: UPSTREAM: <carry>: Does not prevent pod creation because of no nodes reason when it runs under the regular cluster #756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
openshift-merge-robot merged 2 commits into from
May 27, 2021
Merged

Bug 1961925: UPSTREAM: <carry>: Does not prevent pod creation because of no nodes reason when it runs under the regular cluster #756

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
384cbdd
UPSTREAM: <carry>: Does not prevent pod creation because of no nodes …
May 19, 2021
374f6f0
UPSTREAM: <carry>: vendor config listers and informers
May 25, 2021
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
57 changes: 42 additions & 15 deletions openshift-kube-apiserver/admission/autoscaling/managementcpusoverride/admission.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,10 @@ import (
"strings"
"time"

configv1 "github.com/openshift/api/config/v1"
configv1informer "github.com/openshift/client-go/config/informers/externalversions/config/v1"
configv1listers "github.com/openshift/client-go/config/listers/config/v1"

corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
Expand Down Expand Up @@ -44,12 +48,15 @@ const (
namespaceAllowedAnnotation = "workload.openshift.io/allowed"
// workloadAdmissionWarning contains the admission warning annotation key
workloadAdmissionWarning = "workload.openshift.io/warning"
// infraClusterName contains the name of the cluster infrastructure resource
infraClusterName = "cluster"
)

var _ = initializer.WantsExternalKubeInformerFactory(&managementCPUsOverride{})
var _ = initializer.WantsExternalKubeClientSet(&managementCPUsOverride{})
var _ = admission.MutationInterface(&managementCPUsOverride{})
var _ = admission.ValidationInterface(&managementCPUsOverride{})
var _ = WantsInfraInformer(&managementCPUsOverride{})

func Register(plugins *admission.Plugins) {
plugins.Register(PluginName,
Expand All @@ -76,11 +83,13 @@ func Register(plugins *admission.Plugins) {
// 4. The CPU request deletion will not change the pod QoS class
type managementCPUsOverride struct {
*admission.Handler
client kubernetes.Interface
nsLister corev1listers.NamespaceLister
nsListerSynced func() bool
nodeLister corev1listers.NodeLister
nodeListSynced func() bool
client kubernetes.Interface
nsLister corev1listers.NamespaceLister
nsListerSynced func() bool
nodeLister corev1listers.NodeLister
nodeListSynced func() bool
infraConfigLister configv1listers.InfrastructureLister
infraConfigListSynced func() bool
}

func (a *managementCPUsOverride) SetExternalKubeInformerFactory(kubeInformers informers.SharedInformerFactory) {
Expand All @@ -95,6 +104,11 @@ func (a *managementCPUsOverride) SetExternalKubeClientSet(client kubernetes.Inte
a.client = client
}

func (a *managementCPUsOverride) SetInfraInformer(informer configv1informer.InfrastructureInformer) {
a.infraConfigLister = informer.Lister()
a.infraConfigListSynced = informer.Informer().HasSynced
}

func (a *managementCPUsOverride) ValidateInitialization() error {
if a.client == nil {
return fmt.Errorf("%s plugin needs a kubernetes client", PluginName)
Expand All @@ -111,6 +125,12 @@ func (a *managementCPUsOverride) ValidateInitialization() error {
if a.nodeListSynced == nil {
return fmt.Errorf("%s plugin needs a node lister synced", PluginName)
}
if a.infraConfigLister == nil {
return fmt.Errorf("%s did not get a config infrastructure lister", PluginName)
}
if a.infraConfigListSynced == nil {
return fmt.Errorf("%s plugin needs a config infrastructure lister synced", PluginName)
}
return nil
}

Expand Down Expand Up @@ -152,26 +172,33 @@ func (a *managementCPUsOverride) Admit(ctx context.Context, attr admission.Attri
}

if !a.waitForSyncedStore(time.After(timeToWaitForCacheSync)) {
return admission.NewForbidden(attr, fmt.Errorf("%s node and namespace caches not synchronized", PluginName))
return admission.NewForbidden(attr, fmt.Errorf("%s node or namespace or infra config cache not synchronized", PluginName))
}

nodes, err := a.nodeLister.List(labels.Everything())
clusterInfra, err := a.infraConfigLister.Get(infraClusterName)

@sttts sttts May 25, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

can one delete the infra resource? Would that brick the cluster?

@cynepco3hahue cynepco3hahue May 25, 2021

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

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

@dhellmann @browsell Does it expected that an admin can delete the infrastructure object? What additional components in a cluster relay on it?

@deads2k deads2k May 26, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

@sttts if it's deletable, it's a bug. Admission shoudl be coded to prevent deletion of config.openshift.io objects.

@dhellmann dhellmann May 26, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Lots of things rely on that resource. It's entirely possible that someone could delete it. There is also a period of time during bootstrapping where it won't exist yet. So, yes, we need to cope with it not existing and assume a default. Unfortunately, the default won't work for single node because it won't enable partitioning.

That race condition makes me think we need something other than an API resource to turn the feature on, since we need all partitioning annotations processed the same way from the beginning of the life of the cluster. I'm not sure what options we have. Elsewhere I would say use an environment variable or config file. Are those options in the API server @stts & @deads2k ?

@dhellmann dhellmann May 26, 2021

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Nevermind, I misunderstood some things about the ordering during bootstrapping. It should be safe to assume the infrastructure resource exists when it's safe to create regular pods through the API.

if err != nil {
return admission.NewForbidden(attr, err) // can happen due to informer latency
}

// only pods that possible to run without any nodes is static pods, but we skip mutation of static pods in the beginning
if len(nodes) == 0 {
return admission.NewForbidden(attr, fmt.Errorf("%s the cluster does not have any nodes", PluginName))
}

// not the SNO cluster, skip mutation
// TODO: currently we supports only SNO use case because we have not yet worked out the best approach to determining whether the feature
// should be on or off in a multi-node cluster, and computing that state incorrectly could lead to breaking running clusters.
if len(nodes) > 1 {
pod.Annotations[workloadAdmissionWarning] = "only clusters with a single node are supported"
if clusterInfra.Status.InfrastructureTopology != configv1.SingleReplicaTopologyMode ||
clusterInfra.Status.ControlPlaneTopology != configv1.SingleReplicaTopologyMode {
pod.Annotations[workloadAdmissionWarning] = "only single-node clusters support workload partitioning"
return nil
}

nodes, err := a.nodeLister.List(labels.Everything())
if err != nil {
return admission.NewForbidden(attr, err) // can happen due to informer latency
}

// we still need to have nodes under the cluster to decide if the management resource enabled or not
if len(nodes) == 0 {
return admission.NewForbidden(attr, fmt.Errorf("%s the cluster does not have any nodes", PluginName))
}

// probably the workload feature disabled, because some of cluster nodes do not have workload resource
if err := isManagementResourceAvailableForAllNodes(nodes, workloadType); err != nil {
pod.Annotations[workloadAdmissionWarning] = err.Error()
Expand Down Expand Up @@ -260,7 +287,7 @@ func (a *managementCPUsOverride) getPodNamespace(attr admission.Attributes) (*co
}

func (a *managementCPUsOverride) waitForSyncedStore(timeout <-chan time.Time) bool {
for !a.nsListerSynced() || !a.nodeListSynced() {
for !a.nsListerSynced() || !a.nodeListSynced() || !a.infraConfigListSynced() {
select {
case <-time.After(100 * time.Millisecond):
case <-timeout:
Expand Down
73 changes: 61 additions & 12 deletions openshift-kube-apiserver/admission/autoscaling/managementcpusoverride/admission_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,9 @@ import (
"strings"
"testing"

configv1 "github.com/openshift/api/config/v1"
configv1listers "github.com/openshift/client-go/config/listers/config/v1"

corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand All @@ -28,14 +31,16 @@ const (
managedCapacityLabel = "management.workload.openshift.io/cores"
)

func getMockManagementCPUsOverride(namespace *corev1.Namespace, nodes []*corev1.Node) (*managementCPUsOverride, error) {
func getMockManagementCPUsOverride(namespace *corev1.Namespace, nodes []*corev1.Node, infra *configv1.Infrastructure) (*managementCPUsOverride, error) {
m := &managementCPUsOverride{
Handler: admission.NewHandler(admission.Create),
client: &fake.Clientset{},
nsLister: fakeNamespaceLister(namespace),
nsListerSynced: func() bool { return true },
nodeLister: fakeNodeLister(nodes),
nodeListSynced: func() bool { return true },
Handler: admission.NewHandler(admission.Create),
client: &fake.Clientset{},
nsLister: fakeNamespaceLister(namespace),
nsListerSynced: func() bool { return true },
nodeLister: fakeNodeLister(nodes),
nodeListSynced: func() bool { return true },
infraConfigLister: fakeInfraConfigLister(infra),
infraConfigListSynced: func() bool { return true },
}
if err := m.ValidateInitialization(); err != nil {
return nil, err
Expand All @@ -58,12 +63,21 @@ func fakeNodeLister(nodes []*corev1.Node) corev1listers.NodeLister {
return corev1listers.NewNodeLister(indexer)
}

func fakeInfraConfigLister(infra *configv1.Infrastructure) configv1listers.InfrastructureLister {
indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
if infra != nil {
_ = indexer.Add(infra)
}
return configv1listers.NewInfrastructureLister(indexer)
}

func TestAdmit(t *testing.T) {
tests := []struct {
name string
pod *kapi.Pod
namespace *corev1.Namespace
nodes []*corev1.Node
infra *configv1.Infrastructure
expectedCpuRequest resource.Quantity
expectedAnnotations map[string]string
expectedError error
Expand All @@ -74,13 +88,15 @@ func TestAdmit(t *testing.T) {
expectedCpuRequest: resource.MustParse("250m"),
namespace: testNamespace(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
expectedError: fmt.Errorf("the pod namespace %q does not allow the workload type management", "namespace"),
},
{
name: "should ignore pods that do not have managed annotation",
pod: testPod("500m", "250m", "500Mi", "250Mi"),
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
infra: testClusterSNOInfra(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
},
{
Expand All @@ -98,6 +114,7 @@ func TestAdmit(t *testing.T) {
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
expectedError: fmt.Errorf("the pod can not have more than one workload annotations"),
},
{
Expand All @@ -114,6 +131,7 @@ func TestAdmit(t *testing.T) {
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
expectedError: fmt.Errorf("the workload annotation key should have format %s<workload_type>", podWorkloadTargetAnnotationPrefix),
},
{
Expand All @@ -130,6 +148,7 @@ func TestAdmit(t *testing.T) {
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
expectedError: fmt.Errorf(`failed to get workload annotation effect: failed to parse "{" annotation value: unexpected end of JSON input`),
},
{
Expand All @@ -146,6 +165,7 @@ func TestAdmit(t *testing.T) {
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
expectedError: fmt.Errorf(`failed to get workload annotation effect: the workload annotation value map["test":"test"] does not have "effect" key`),
},
{
Expand All @@ -159,6 +179,7 @@ func TestAdmit(t *testing.T) {
fmt.Sprintf("%s%s", podWorkloadTargetAnnotationPrefix, workloadTypeManagement): fmt.Sprintf(`{"%s":"%s"}`, podWorkloadAnnotationEffect, workloadEffectPreferredDuringScheduling),
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should update workload CPU annotations for the best-effort pod with managed annotation",
Expand All @@ -171,6 +192,7 @@ func TestAdmit(t *testing.T) {
fmt.Sprintf("%s%s", podWorkloadTargetAnnotationPrefix, workloadTypeManagement): fmt.Sprintf(`{"%s":"%s"}`, podWorkloadAnnotationEffect, workloadEffectPreferredDuringScheduling),
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should skip static pod mutation",
Expand All @@ -182,6 +204,7 @@ func TestAdmit(t *testing.T) {
kubetypes.ConfigSourceAnnotationKey: kubetypes.FileSource,
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should ignore guaranteed pod",
Expand All @@ -192,6 +215,7 @@ func TestAdmit(t *testing.T) {
workloadAdmissionWarning: "skip pod CPUs requests modifications because it has guaranteed QoS class",
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should ignore pod when removing the CPU request will change the pod QoS class to guaranteed",
Expand All @@ -202,6 +226,7 @@ func TestAdmit(t *testing.T) {
workloadAdmissionWarning: fmt.Sprintf("skip pod CPUs requests modifications because it will change the pod QoS class from %s to %s", corev1.PodQOSBurstable, corev1.PodQOSGuaranteed),
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should ignore pod when removing the CPU request will change the pod QoS class to best-effort",
Expand All @@ -212,16 +237,26 @@ func TestAdmit(t *testing.T) {
workloadAdmissionWarning: fmt.Sprintf("skip pod CPUs requests modifications because it will change the pod QoS class from %s to %s", corev1.PodQOSBurstable, corev1.PodQOSBestEffort),
},
nodes: []*corev1.Node{testNodeWithManagementResource()},
infra: testClusterSNOInfra(),
},
{
name: "should not mutate the pod when the cluster has more than one node",
name: "should not mutate the pod, when it runs under the regular(non SNO) cluster",
pod: testManagedPod("500m", "250m", "500Mi", "250Mi"),
expectedCpuRequest: resource.MustParse("250m"),
namespace: testManagedNamespace(),
expectedAnnotations: map[string]string{
workloadAdmissionWarning: "only clusters with a single node are supported",
workloadAdmissionWarning: "only single-node clusters support workload partitioning",
},
nodes: []*corev1.Node{},
infra: &configv1.Infrastructure{
ObjectMeta: metav1.ObjectMeta{
Name: infraClusterName,
},
Status: configv1.InfrastructureStatus{
ControlPlaneTopology: configv1.HighlyAvailableTopologyMode,
InfrastructureTopology: configv1.SingleReplicaTopologyMode,
},
},
nodes: []*corev1.Node{testNodeWithManagementResource(), testNode()},
},
{
name: "should not mutate the pod when at least one node does not have management resources",
Expand All @@ -232,6 +267,7 @@ func TestAdmit(t *testing.T) {
workloadAdmissionWarning: fmt.Sprintf("the node %q does not have resource %q", "node", managedCapacityLabel),
},
nodes: []*corev1.Node{testNode()},
infra: testClusterSNOInfra(),
},
{
name: "should return admission error when the cluster does not have any nodes",
Expand All @@ -240,12 +276,13 @@ func TestAdmit(t *testing.T) {
namespace: testManagedNamespace(),
nodes: []*corev1.Node{},
expectedError: fmt.Errorf("the cluster does not have any nodes"),
infra: testClusterSNOInfra(),
},
}

for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
m, err := getMockManagementCPUsOverride(test.namespace, test.nodes)
m, err := getMockManagementCPUsOverride(test.namespace, test.nodes, test.infra)
if err != nil {
t.Fatalf("%s: failed to get mock managementCPUsOverride: %v", test.name, err)
}
Expand Down Expand Up @@ -470,7 +507,7 @@ func TestValidate(t *testing.T) {

for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
m, err := getMockManagementCPUsOverride(test.namespace, test.nodes)
m, err := getMockManagementCPUsOverride(test.namespace, test.nodes, nil)
if err != nil {
t.Fatalf("%s: failed to get mock managementCPUsOverride: %v", test.name, err)
}
Expand Down Expand Up @@ -638,3 +675,15 @@ func testNodeWithManagementResource() *corev1.Node {
},
}
}

func testClusterSNOInfra() *configv1.Infrastructure {
return &configv1.Infrastructure{
ObjectMeta: metav1.ObjectMeta{
Name: infraClusterName,
},
Status: configv1.InfrastructureStatus{
ControlPlaneTopology: configv1.SingleReplicaTopologyMode,
InfrastructureTopology: configv1.SingleReplicaTopologyMode,
},
}
}
28 changes: 28 additions & 0 deletions openshift-kube-apiserver/admission/autoscaling/managementcpusoverride/initializers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
package managementcpusoverride

import (
"k8s.io/apiserver/pkg/admission"

configv1informer "github.com/openshift/client-go/config/informers/externalversions/config/v1"
)

func NewInitializer(infraInformer configv1informer.InfrastructureInformer) admission.PluginInitializer {
return &localInitializer{infraInformer: infraInformer}
}

type WantsInfraInformer interface {
SetInfraInformer(informer configv1informer.InfrastructureInformer)
admission.InitializationValidator
}

type localInitializer struct {
infraInformer configv1informer.InfrastructureInformer
}

// Initialize will check the initialization interfaces implemented by each plugin
// and provide the appropriate initialization data
func (i *localInitializer) Initialize(plugin admission.Interface) {
if wants, ok := plugin.(WantsInfraInformer); ok {
wants.SetInfraInformer(i.infraInformer)
}
}
Loading