Skip to content

Commit c310d0f

Browse files
BobVanBBobVanB
committed
fix: dynamic namespaces
1 parent ade0ba0 commit c310d0f

7 files changed

Lines changed: 445 additions & 32 deletions

File tree

‎cmd/manager/main.go‎

Lines changed: 127 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import (
1212
"net/http/pprof"
1313
"os"
1414
"path/filepath"
15+
"sort"
1516
"strconv"
1617
"strings"
1718
"time"
@@ -91,6 +92,7 @@ import (
9192
lsvalidation "github.com/elastic/cloud-on-k8s/v3/pkg/controller/logstash/validation"
9293
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/maps"
9394
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/packageregistry"
95+
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/namespacefilter"
9496
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/remotecluster"
9597
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/stackconfigpolicy"
9698
"github.com/elastic/cloud-on-k8s/v3/pkg/controller/webhook"
@@ -580,6 +582,13 @@ func startOperator(ctx context.Context) error {
580582
// also set up the v1beta1 scheme, used by the v1beta1 webhook
581583
controllerscheme.SetupV1beta1Scheme()
582584

585+
// Setup a client to query namespaces and set operator uuid config map.
586+
clientset, err := kubernetes.NewForConfig(cfg)
587+
if err != nil {
588+
log.Error(err, "Failed to create Kubernetes client")
589+
return err
590+
}
591+
583592
// Create a new Cmd to provide shared dependencies and start components
584593
opts := ctrl.Options{
585594
Scheme: clientgoscheme.Scheme,
@@ -597,22 +606,57 @@ func startOperator(ctx context.Context) error {
597606
log.Error(err, "Failed to parse managed namespaces flag")
598607
return err
599608
}
600-
switch {
601-
case len(managedNamespaces) == 0:
602-
log.Info("Operator configured to manage all namespaces")
603-
case len(managedNamespaces) == 1 && managedNamespaces[0] == operatorNamespace:
604-
log.Info("Operator configured to manage a single namespace", "namespace", managedNamespaces[0], "operator_namespace", operatorNamespace)
605609

606-
default:
607-
log.Info("Operator configured to manage multiple namespaces", "namespaces", managedNamespaces, "operator_namespace", operatorNamespace)
610+
namespaceLabelSelectorStr := viper.GetString(operator.NamespaceLabelSelectorFlag)
611+
configuredNamespaces := managedNamespaces
612+
managedNamespaces, namespaceLabelSelector, err := resolveManagedNamespaces(ctx, clientset, configuredNamespaces, namespaceLabelSelectorStr)
613+
if err != nil {
614+
log.Error(err, "Failed to resolve managed namespaces from selector", "selector", namespaceLabelSelectorStr)
615+
return err
616+
}
617+
618+
var namespaceFilter *operator.NamespaceFilter
619+
if namespaceLabelSelector != nil {
620+
namespaceFilter, err = operator.NewNamespaceFilter(namespaceLabelSelector, configuredNamespaces, managedNamespaces)
621+
if err != nil {
622+
log.Error(err, "Failed to initialize namespace filter")
623+
return err
624+
}
625+
}
626+
627+
restrictedScope := len(configuredNamespaces) > 0
628+
if !restrictedScope {
629+
if namespaceLabelSelector == nil {
630+
log.Info("Operator configured to manage all namespaces")
631+
} else {
632+
log.Info("Operator configured for dynamic namespace label filtering", "selector", namespaceLabelSelector, "initial_managed_namespaces", managedNamespaces)
633+
}
634+
} else {
608635
// The managed cache should always include the operator namespace so that we can work with operator-internal resources.
609-
managedNamespaces = append(managedNamespaces, operatorNamespace)
636+
managedNamespaces = appendIfMissing(configuredNamespaces, operatorNamespace)
637+
managedNamespaces = uniqueNamespaces(managedNamespaces)
638+
sort.Strings(managedNamespaces)
639+
640+
switch {
641+
case namespaceLabelSelector != nil && len(managedNamespaces) == 1 && managedNamespaces[0] == operatorNamespace:
642+
log.Info("Namespace label selector matched no namespaces; limiting cache to operator namespace", "operator_namespace", operatorNamespace, "selector", namespaceLabelSelectorStr)
643+
case len(managedNamespaces) == 1:
644+
log.Info("Operator configured to manage a single namespace", "namespace", managedNamespaces[0], "operator_namespace", operatorNamespace)
645+
default:
646+
log.Info("Operator configured to manage multiple namespaces", "namespaces", managedNamespaces, "operator_namespace", operatorNamespace)
647+
}
648+
649+
if namespaceLabelSelector != nil {
650+
log.Info("Namespace label selector configured", "selector", namespaceLabelSelector)
651+
}
610652
}
611653

612-
// implicitly allows watching cluster-scoped resources (e.g. storage classes)
613-
opts.Cache = cache.Options{DefaultNamespaces: map[string]cache.Config{}}
614-
for _, ns := range managedNamespaces {
615-
opts.Cache.DefaultNamespaces[ns] = cache.Config{}
654+
if restrictedScope {
655+
// implicitly allows watching cluster-scoped resources (e.g. storage classes)
656+
opts.Cache = cache.Options{DefaultNamespaces: map[string]cache.Config{}}
657+
for _, ns := range managedNamespaces {
658+
opts.Cache.DefaultNamespaces[ns] = cache.Config{}
659+
}
616660
}
617661

618662
// only expose prometheus metrics if provided a non-zero port
@@ -676,13 +720,6 @@ func startOperator(ctx context.Context) error {
676720
return err
677721
}
678722

679-
// Setup a client to set the operator uuid config map
680-
clientset, err := kubernetes.NewForConfig(cfg)
681-
if err != nil {
682-
log.Error(err, "Failed to create Kubernetes client")
683-
return err
684-
}
685-
686723
distributionChannel := viper.GetString(operator.DistributionChannelFlag)
687724
operatorInfo, err := about.GetOperatorInfo(clientset, operatorNamespace, distributionChannel)
688725
if err != nil {
@@ -725,19 +762,6 @@ func startOperator(ctx context.Context) error {
725762
return err
726763
}
727764

728-
// Parse namespace label selector if provided
729-
var namespaceLabelSelector *metav1.LabelSelector
730-
namespaceLabelSelectorStr := viper.GetString(operator.NamespaceLabelSelectorFlag)
731-
if namespaceLabelSelectorStr != "" {
732-
selector, err := metav1.ParseToLabelSelector(namespaceLabelSelectorStr)
733-
if err != nil {
734-
log.Error(err, "Failed to parse namespace label selector", "selector", namespaceLabelSelectorStr)
735-
return err
736-
}
737-
namespaceLabelSelector = selector
738-
log.Info("Namespace label selector configured", "selector", namespaceLabelSelector)
739-
}
740-
741765
params := operator.Parameters{
742766
Dialer: dialer,
743767
ElasticsearchObservationInterval: viper.GetDuration(operator.ElasticsearchObservationIntervalFlag),
@@ -761,6 +785,7 @@ func startOperator(ctx context.Context) error {
761785
ValidateStorageClass: viper.GetBool(operator.ValidateStorageClassFlag),
762786
Tracer: tracer,
763787
NamespaceLabelSelector: namespaceLabelSelector,
788+
NamespaceFilter: namespaceFilter,
764789
}
765790

766791
if viper.GetBool(operator.EnableWebhookFlag) {
@@ -830,6 +855,75 @@ func readOptionalCA(caDir string) (*certificates.CA, error) {
830855
return certificates.BuildCAFromFile(caDir)
831856
}
832857

858+
func resolveManagedNamespaces(
859+
ctx context.Context,
860+
clientset kubernetes.Interface,
861+
configuredNamespaces []string,
862+
namespaceLabelSelectorStr string,
863+
) ([]string, *metav1.LabelSelector, error) {
864+
if namespaceLabelSelectorStr == "" {
865+
return configuredNamespaces, nil, nil
866+
}
867+
868+
namespaceLabelSelector, err := metav1.ParseToLabelSelector(namespaceLabelSelectorStr)
869+
if err != nil {
870+
return nil, nil, err
871+
}
872+
873+
namespaceList, err := clientset.CoreV1().Namespaces().List(ctx, metav1.ListOptions{LabelSelector: namespaceLabelSelectorStr})
874+
if err != nil {
875+
return nil, nil, err
876+
}
877+
878+
selectorNamespaces := make([]string, 0, len(namespaceList.Items))
879+
for _, namespace := range namespaceList.Items {
880+
selectorNamespaces = append(selectorNamespaces, namespace.Name)
881+
}
882+
883+
if len(configuredNamespaces) == 0 {
884+
return selectorNamespaces, namespaceLabelSelector, nil
885+
}
886+
887+
selectorNamespacesSet := make(map[string]struct{}, len(selectorNamespaces))
888+
for _, namespace := range selectorNamespaces {
889+
selectorNamespacesSet[namespace] = struct{}{}
890+
}
891+
892+
filteredNamespaces := make([]string, 0, len(configuredNamespaces))
893+
for _, namespace := range configuredNamespaces {
894+
if _, exists := selectorNamespacesSet[namespace]; exists {
895+
filteredNamespaces = append(filteredNamespaces, namespace)
896+
}
897+
}
898+
899+
return filteredNamespaces, namespaceLabelSelector, nil
900+
}
901+
902+
func appendIfMissing(namespaces []string, namespace string) []string {
903+
for _, existing := range namespaces {
904+
if existing == namespace {
905+
return namespaces
906+
}
907+
}
908+
return append(namespaces, namespace)
909+
}
910+
911+
func uniqueNamespaces(namespaces []string) []string {
912+
unique := make([]string, 0, len(namespaces))
913+
seen := make(map[string]struct{}, len(namespaces))
914+
for _, namespace := range namespaces {
915+
if namespace == "" {
916+
continue
917+
}
918+
if _, alreadySeen := seen[namespace]; alreadySeen {
919+
continue
920+
}
921+
seen[namespace] = struct{}{}
922+
unique = append(unique, namespace)
923+
}
924+
return unique
925+
}
926+
833927
// asyncTasks schedules some tasks to be started when this instance of the operator is elected
834928
func asyncTasks(
835929
ctx context.Context,
@@ -936,6 +1030,7 @@ func registerControllers(mgr manager.Manager, params operator.Parameters, access
9361030
name string
9371031
registerFunc func(manager.Manager, operator.Parameters) error
9381032
}{
1033+
{name: "NamespaceFilter", registerFunc: namespacefilter.Add},
9391034
{name: "APMServer", registerFunc: apmserver.Add},
9401035
{name: "Elasticsearch", registerFunc: elasticsearch.Add},
9411036
{name: "ElasticsearchAutoscaling", registerFunc: autoscaling.Add},

‎cmd/manager/main_test.go‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import (
1717
"k8s.io/apimachinery/pkg/types"
1818
"k8s.io/client-go/discovery"
1919
"k8s.io/client-go/kubernetes"
20+
"k8s.io/client-go/kubernetes/fake"
2021
"sigs.k8s.io/controller-runtime/pkg/client"
2122
logf "sigs.k8s.io/controller-runtime/pkg/log"
2223

@@ -337,3 +338,64 @@ func newFakeK8sClientsetWithDiscovery(resources []*metav1.APIResourceList, disco
337338
}
338339
return client
339340
}
341+
342+
func Test_resolveManagedNamespaces(t *testing.T) {
343+
fakeClientset := fake.NewSimpleClientset(
344+
&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "prod-a", Labels: map[string]string{"env": "prod"}}},
345+
&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "dev-a", Labels: map[string]string{"env": "dev"}}},
346+
&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "prod-b", Labels: map[string]string{"env": "prod"}}},
347+
)
348+
349+
tests := []struct {
350+
name string
351+
configuredNamespaces []string
352+
selector string
353+
expectedNamespaces []string
354+
expectedSelectorNil bool
355+
expectErr bool
356+
}{
357+
{
358+
name: "no selector returns configured namespaces",
359+
configuredNamespaces: []string{"prod-a", "dev-a"},
360+
selector: "",
361+
expectedNamespaces: []string{"prod-a", "dev-a"},
362+
expectedSelectorNil: true,
363+
expectErr: false,
364+
},
365+
{
366+
name: "selector resolves namespaces when none configured",
367+
configuredNamespaces: nil,
368+
selector: "env=prod",
369+
expectedNamespaces: []string{"prod-a", "prod-b"},
370+
expectedSelectorNil: false,
371+
expectErr: false,
372+
},
373+
{
374+
name: "selector filters configured namespaces by intersection",
375+
configuredNamespaces: []string{"dev-a", "prod-b"},
376+
selector: "env=prod",
377+
expectedNamespaces: []string{"prod-b"},
378+
expectedSelectorNil: false,
379+
expectErr: false,
380+
},
381+
}
382+
383+
for _, tt := range tests {
384+
t.Run(tt.name, func(t *testing.T) {
385+
namespaces, selector, err := resolveManagedNamespaces(context.Background(), fakeClientset, tt.configuredNamespaces, tt.selector)
386+
387+
if tt.expectErr {
388+
require.Error(t, err)
389+
return
390+
}
391+
392+
require.NoError(t, err)
393+
require.Equal(t, tt.expectedNamespaces, namespaces)
394+
if tt.expectedSelectorNil {
395+
require.Nil(t, selector)
396+
} else {
397+
require.NotNil(t, selector)
398+
}
399+
})
400+
}
401+
}

0 commit comments

Comments
 (0)