Skip to content
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

[Multicluster] Fix cache options for controller Manager #6150

Merged
merged 2 commits into from
Mar 27, 2024
Merged

[Multicluster] Fix cache options for controller Manager #6150

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f7e8260
[Multicluster] Fix cache options for controller Manager
antoninbas Mar 26, 2024
6d251b2
Fix panic in admission webhook
antoninbas Mar 27, 2024
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
5 changes: 0 additions & 5 deletions multicluster/cmd/multicluster-controller/clusterset_webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,3 @@ func (v *clusterSetValidator) Handle(ctx context.Context, req admission.Request)
}
return admission.Allowed("")
}

func (v *clusterSetValidator) InjectDecoder(d *admission.Decoder) error {
v.decoder = d
return nil
}
3 changes: 1 addition & 2 deletions multicluster/cmd/multicluster-controller/clusterset_webhook_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,6 @@ func TestWebhookClusterSetEvents(t *testing.T) {
}

decoder := admission.NewDecoder(common.TestScheme)

for _, tt := range tests {
objects := []client.Object{}
if tt.existingClusterSet != nil {
Expand All @@ -193,10 +192,10 @@ func TestWebhookClusterSetEvents(t *testing.T) {
fakeClient := fake.NewClientBuilder().WithScheme(common.TestScheme).WithObjects(objects...).Build()
clusterSetWebhookUnderTest = &clusterSetValidator{
Client: fakeClient,
decoder: decoder,
namespace: "mcs1",
role: tt.role,
}
clusterSetWebhookUnderTest.InjectDecoder(decoder)

t.Run(tt.name, func(t *testing.T) {
response := clusterSetWebhookUnderTest.Handle(context.Background(), tt.req)
Expand Down
58 changes: 35 additions & 23 deletions multicluster/cmd/multicluster-controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import (
apiextensionclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
genericoptions "k8s.io/apiserver/pkg/server/options"
clientset "k8s.io/client-go/kubernetes"
Expand Down Expand Up @@ -139,6 +138,15 @@ func getWebhookLabel(isLeader bool, controllerNs string) *metav1.LabelSelector {
func setupManagerAndCertController(isLeader bool, o *Options) (manager.Manager, error) {
ctrl.SetLogger(klog.NewKlogr())

podNamespace := env.GetPodNamespace()

var caConfig *certificate.CAConfig
if isLeader {
caConfig = getCaConfig(isLeader, podNamespace)
} else {
caConfig = getCaConfig(isLeader, "")
}

// build up cert controller to manage certificate for MC Controller
k8sConfig := ctrl.GetConfigOrDie()
k8sConfig.QPS = common.ResourceExchangeQPS
Expand All @@ -149,40 +157,48 @@ func setupManagerAndCertController(isLeader bool, o *Options) (manager.Manager,
}

secureServing := genericoptions.NewSecureServingOptions().WithLoopback()
caCertController, err := certificate.ApplyServerCert(o.SelfSignedCert, client, aggregatorClient, apiExtensionClient,
secureServing, getCaConfig(isLeader, o.Namespace))
caCertController, err := certificate.ApplyServerCert(o.SelfSignedCert, client, aggregatorClient, apiExtensionClient, secureServing, caConfig)
if err != nil {
return nil, fmt.Errorf("error applying server cert: %v", err)
}
if err := caCertController.RunOnce(context.TODO()); err != nil {
return nil, err
}

options := o.options
if o.SelfSignedCert {
o.options.Metrics.CertDir = selfSignedCertDir
options.Metrics.CertDir = selfSignedCertDir
o.WebhookConfig.CertDir = selfSignedCertDir
} else {
o.options.Metrics.CertDir = certDir
options.Metrics.CertDir = certDir
o.WebhookConfig.CertDir = certDir
}
o.options.WebhookServer = webhook.NewServer(webhook.Options{
options.WebhookServer = webhook.NewServer(webhook.Options{
Port: *o.WebhookConfig.Port,
Host: o.WebhookConfig.Host,
CertDir: o.WebhookConfig.CertDir,
})

namespaceFieldSelector := fields.SelectorFromSet(fields.Set{"metadata.namespace": env.GetPodNamespace()})
o.options.Cache.DefaultFieldSelector = namespaceFieldSelector
o.options.Cache.ByObject = map[controllerruntimeclient.Object]cache.ByObject{
&mcv1alpha1.Gateway{}: {
Field: namespaceFieldSelector,
},
&mcv1alpha2.ClusterSet{}: {
Field: namespaceFieldSelector,
},
&mcv1alpha1.MemberClusterAnnounce{}: {
Field: namespaceFieldSelector,
},
cacheOptions := &options.Cache
if isLeader {
// For the leader, restrict the cache to the controller's Namespace.
cacheOptions.DefaultNamespaces = map[string]cache.Config{
podNamespace: {},
}
} else {
// For a member, restict the cache to the controller's Namespace for the following objects.
Copy link
Contributor Author

Choose a reason for hiding this comment

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

@luolanzone I am not entirely sure how these objects are determined.
However, I removed mcv1alpha1.MemberClusterAnnounce, which doesn't exist in the member cluster.

Copy link
Contributor

Choose a reason for hiding this comment

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

Yeah, it should be removed if we separate the filter for the leader and member.

cacheOptions.ByObject = map[controllerruntimeclient.Object]cache.ByObject{
&mcv1alpha1.Gateway{}: {
Namespaces: map[string]cache.Config{
podNamespace: {},
},
},
&mcv1alpha2.ClusterSet{}: {
Namespaces: map[string]cache.Config{
podNamespace: {},
},
},
}
}

// EndpointSlice is enabled in AntreaProxy by default since v1.11, so Antrea MC
Expand All @@ -206,11 +222,7 @@ func setupManagerAndCertController(isLeader bool, o *Options) (manager.Manager,
}
o.ClusterCalimCRDAvailable = clusterClaimCRDAvailable

mgr, err := ctrl.NewManager(k8sConfig, manager.Options{
Scheme: o.options.Scheme,
Metrics: o.options.Metrics,
HealthProbeBindAddress: o.options.HealthProbeBindAddress,
})
mgr, err := ctrl.NewManager(k8sConfig, options)
if err != nil {
return nil, fmt.Errorf("error creating manager: %v", err)
}
Expand Down
5 changes: 0 additions & 5 deletions multicluster/cmd/multicluster-controller/gateway_webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,3 @@ func (v *gatewayValidator) Handle(ctx context.Context, req admission.Request) ad
}
return admission.Allowed("")
}

func (v *gatewayValidator) InjectDecoder(d *admission.Decoder) error {
v.decoder = d
return nil
}
5 changes: 3 additions & 2 deletions multicluster/cmd/multicluster-controller/gateway_webhook_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,8 +180,9 @@ func TestWebhookGatewayEvents(t *testing.T) {
}
gatewayWebhookUnderTest = &gatewayValidator{
Client: fakeClient,
namespace: "default"}
gatewayWebhookUnderTest.InjectDecoder(decoder)
decoder: decoder,
namespace: "default",
}

t.Run(tt.name, func(t *testing.T) {
response := gatewayWebhookUnderTest.Handle(context.Background(), tt.req)
Expand Down
14 changes: 9 additions & 5 deletions multicluster/cmd/multicluster-controller/leader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/webhook"
"sigs.k8s.io/controller-runtime/pkg/webhook/admission"

multiclusterv1alpha1 "antrea.io/antrea/multicluster/apis/multicluster/v1alpha1"
"antrea.io/antrea/multicluster/controllers/multicluster/leader"
Expand Down Expand Up @@ -50,9 +51,7 @@ func newLeaderCommand() *cobra.Command {
}

func runLeader(o *Options) error {
// on the leader we want the reconciler to run for a given Namespace instead of cluster scope
podNamespace := env.GetPodNamespace()
o.Namespace = podNamespace
stopCh := signals.RegisterSignalHandlers()

mgr, err := setupManagerAndCertControllerFunc(true, o)
Expand All @@ -76,14 +75,19 @@ func runLeader(o *Options) error {
hookServer.Register("/validate-multicluster-crd-antrea-io-v1alpha1-memberclusterannounce",
&webhook.Admission{Handler: &memberClusterAnnounceValidator{
Client: noCachedClient,
namespace: podNamespace}})
decoder: admission.NewDecoder(mgr.GetScheme()),
namespace: podNamespace,
}},
)

hookServer.Register("/validate-multicluster-crd-antrea-io-v1alpha2-clusterset",
&webhook.Admission{Handler: &clusterSetValidator{
Client: mgr.GetClient(),
decoder: admission.NewDecoder(mgr.GetScheme()),
namespace: env.GetPodNamespace(),
role: leaderRole},
})
role: leaderRole,
}},
)

clusterSetReconciler := leader.NewLeaderClusterSetReconciler(mgrClient, podNamespace,
o.ClusterCalimCRDAvailable, memberClusterStatusManager)
Expand Down
12 changes: 9 additions & 3 deletions multicluster/cmd/multicluster-controller/member.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"k8s.io/klog/v2"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/webhook"
"sigs.k8s.io/controller-runtime/pkg/webhook/admission"

"antrea.io/antrea/multicluster/controllers/multicluster/member"
"antrea.io/antrea/pkg/log"
Expand Down Expand Up @@ -62,14 +63,19 @@ func runMember(o *Options) error {
hookServer.Register("/validate-multicluster-crd-antrea-io-v1alpha1-gateway",
&webhook.Admission{Handler: &gatewayValidator{
Client: mgrClient,
namespace: podNamespace}})
decoder: admission.NewDecoder(mgr.GetScheme()),
namespace: podNamespace,
}},
)

hookServer.Register("/validate-multicluster-crd-antrea-io-v1alpha2-clusterset",
&webhook.Admission{Handler: &clusterSetValidator{
Client: mgrClient,
decoder: admission.NewDecoder(mgr.GetScheme()),
namespace: podNamespace,
role: memberRole},
})
role: memberRole,
}},
)

commonAreaCreationCh := make(chan struct{}, 1)
clusterSetReconciler := member.NewMemberClusterSetReconciler(mgr.GetClient(),
Expand Down
5 changes: 0 additions & 5 deletions multicluster/cmd/multicluster-controller/memberclusterannounce_webhook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,8 +108,3 @@ func (v *memberClusterAnnounceValidator) Handle(ctx context.Context, req admissi
return admission.Allowed("")
}
}

func (v *memberClusterAnnounceValidator) InjectDecoder(d *admission.Decoder) error {
v.decoder = d
return nil
}
5 changes: 3 additions & 2 deletions multicluster/cmd/multicluster-controller/memberclusterannounce_webhook_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,8 +259,9 @@ func TestMemberClusterAnnounceWebhook(t *testing.T) {
}
mcaWebhookUnderTest = &memberClusterAnnounceValidator{
Client: fakeClient,
namespace: "mcs1"}
mcaWebhookUnderTest.InjectDecoder(decoder)
decoder: decoder,
namespace: "mcs1",
}
t.Run(tt.name, func(t *testing.T) {
response := mcaWebhookUnderTest.Handle(context.Background(), tt.req)
assert.Equal(t, tt.isAllowed, response.Allowed)
Expand Down
6 changes: 2 additions & 4 deletions multicluster/cmd/multicluster-controller/options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,8 @@ type Options struct {
// The path of configuration file.
configFile string
SelfSignedCert bool
options ctrl.Options
Namespace string
// options store some base controller Manager options (initialized from the provided config).
options ctrl.Options
// The Service ClusterIP range used in the member cluster.
ServiceCIDR string
// PodCIDRs is the Pod IP address CIDRs of the member cluster.
Expand Down Expand Up @@ -68,14 +68,12 @@ func newOptions() *Options {
func (o *Options) complete(args []string) error {
var err error
o.setDefaults()
options := ctrl.Options{Scheme: scheme}
ctrlConfig := &mcsv1alpha1.MultiClusterConfig{}
if len(o.configFile) > 0 {
klog.InfoS("Loading config", "file", o.configFile)
if err = o.loadConfigFromFile(ctrlConfig); err != nil {
return err
}
o.options = options
if ctrlConfig.ServiceCIDR != "" {
if _, _, err := net.ParseCIDR(ctrlConfig.ServiceCIDR); err != nil {
return fmt.Errorf("failed to parse serviceCIDR, invalid CIDR string %s", ctrlConfig.ServiceCIDR)
Expand Down
5 changes: 0 additions & 5 deletions multicluster/cmd/multicluster-controller/options_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"testing"

"github.com/stretchr/testify/assert"
ctrl "sigs.k8s.io/controller-runtime"
)

func TestComplete(t *testing.T) {
Expand All @@ -34,7 +33,6 @@ func TestComplete(t *testing.T) {
o: Options{
configFile: "./testdata/antrea-mc-config-with-valid-podcidrs.yml",
SelfSignedCert: false,
options: ctrl.Options{},
ServiceCIDR: "",
PodCIDRs: nil,
GatewayIPPrecedence: "",
Expand All @@ -47,7 +45,6 @@ func TestComplete(t *testing.T) {
o: Options{
configFile: "./testdata/antrea-mc-config-with-empty-podcidrs.yml",
SelfSignedCert: false,
options: ctrl.Options{},
ServiceCIDR: "",
PodCIDRs: nil,
GatewayIPPrecedence: "",
Expand All @@ -60,7 +57,6 @@ func TestComplete(t *testing.T) {
o: Options{
configFile: "./testdata/antrea-mc-config-with-invalid-podcidrs.yml",
SelfSignedCert: false,
options: ctrl.Options{},
ServiceCIDR: "10.100.0.0/16",
PodCIDRs: nil,
GatewayIPPrecedence: "",
Expand All @@ -73,7 +69,6 @@ func TestComplete(t *testing.T) {
o: Options{
configFile: "./testdata/antrea-mc-config-with-invalid-endpointiptype.yml",
SelfSignedCert: false,
options: ctrl.Options{},
ServiceCIDR: "10.100.0.0/16",
PodCIDRs: nil,
GatewayIPPrecedence: "",
Expand Down
Loading