admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: clusterrolebinding controller bug

Browse Source 
Signed-off-by: hongming <talonwan@yunify.com>
This commit is contained in:
hongming
2020-05-29 12:49:28 +08:00
parent 1882b20e6e
commit 4e418bbe03
7 changed files with 67 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
cmd/controller-manager/app/controllers.go
View File

@@ -125,9 +125,11 @@ func AddControllers(
csrController := certificatesigningrequest.NewController(client.Kubernetes(), kubernetesInformer, client.Config())
clusterRoleBindingController := clusterrolebinding.NewController(client.Kubernetes(), kubernetesInformer, kubesphereInformer)
clusterRoleBindingController := clusterrolebinding.NewController(client.Kubernetes(),
kubernetesInformer.Rbac().V1().ClusterRoleBindings(), kubernetesInformer.Apps().V1().Deployments(),
kubernetesInformer.Core().V1().Pods(), kubesphereInformer.Iam().V1alpha2().Users())
globalRoleBindingController := globalrolebinding.NewController(client.Kubernetes(), kubernetesInformer, kubesphereInformer, multiClusterEnabled)
globalRoleBindingController := globalrolebinding.NewController(client.Kubernetes(), kubesphereInformer.Iam().V1alpha2().GlobalRoleBindings(), multiClusterEnabled)
clusterController := cluster.NewClusterController(
client.Kubernetes(),

38
pkg/controller/clusterrolebinding/clusterrolebinding_controller.go
View File

@@ -22,7 +22,8 @@ import (
"k8s.io/apimachinery/pkg/api/errors"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
k8sinformers "k8s.io/client-go/informers"
appsv1informers "k8s.io/client-go/informers/apps/v1"
coreinfomers "k8s.io/client-go/informers/core/v1"
rbacv1informers "k8s.io/client-go/informers/rbac/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
@@ -33,7 +34,7 @@ import (
"k8s.io/client-go/util/workqueue"
"k8s.io/klog"
iamv1alpha2 "kubesphere.io/kubesphere/pkg/apis/iam/v1alpha2"
ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
iamv1alpha2informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/iam/v1alpha2"
"kubesphere.io/kubesphere/pkg/models/kubectl"
"time"
)
@@ -47,10 +48,11 @@ const (
)
type Controller struct {
k8sClient kubernetes.Interface
informer rbacv1informers.ClusterRoleBindingInformer
lister rbacv1listers.ClusterRoleBindingLister
synced cache.InformerSynced
k8sClient kubernetes.Interface
clusterRoleBindingInformer rbacv1informers.ClusterRoleBindingInformer
clusterRoleBindingLister rbacv1listers.ClusterRoleBindingLister
clusterRoleBindingSynced cache.InformerSynced
userSynced cache.InformerSynced
// workqueue is a rate limited work queue. This is used to queue work to be
// processed instead of performing it as soon as a change happens. This
// means we can ensure we only process a fixed amount of resources at a
@@ -63,7 +65,7 @@ type Controller struct {
kubectlOperator kubectl.Interface
}
func NewController(k8sClient kubernetes.Interface, k8sInformer k8sinformers.SharedInformerFactory, ksInformer ksinformers.SharedInformerFactory) *Controller {
func NewController(k8sClient kubernetes.Interface, clusterRoleBindingInformer rbacv1informers.ClusterRoleBindingInformer, deploymentInformer appsv1informers.DeploymentInformer, podInformer coreinfomers.PodInformer, userInformer iamv1alpha2informers.UserInformer) *Controller {
// Create event broadcaster
// Add sample-controller types to the default Kubernetes Scheme so Events can be
// logged for sample-controller types.
@@ -73,18 +75,18 @@ func NewController(k8sClient kubernetes.Interface, k8sInformer k8sinformers.Shar
eventBroadcaster.StartLogging(klog.Infof)
eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: k8sClient.CoreV1().Events("")})
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
informer := k8sInformer.Rbac().V1().ClusterRoleBindings()
ctl := &Controller{
k8sClient: k8sClient,
informer: informer,
lister: informer.Lister(),
synced: informer.Informer().HasSynced,
kubectlOperator: kubectl.NewOperator(k8sClient, k8sInformer, ksInformer),
workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "ClusterRoleBinding"),
recorder: recorder,
k8sClient: k8sClient,
clusterRoleBindingInformer: clusterRoleBindingInformer,
clusterRoleBindingLister: clusterRoleBindingInformer.Lister(),
clusterRoleBindingSynced: clusterRoleBindingInformer.Informer().HasSynced,
userSynced: userInformer.Informer().HasSynced,
kubectlOperator: kubectl.NewOperator(k8sClient, deploymentInformer, podInformer, userInformer),
workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "ClusterRoleBinding"),
recorder: recorder,
}
klog.Info("Setting up event handlers")
informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
clusterRoleBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: ctl.enqueueClusterRoleBinding,
UpdateFunc: func(old, new interface{}) {
ctl.enqueueClusterRoleBinding(new)
@@ -105,7 +107,7 @@ func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
// Wait for the caches to be synced before starting workers
klog.Info("Waiting for informer caches to sync")
if ok := cache.WaitForCacheSync(stopCh, c.synced); !ok {
if ok := cache.WaitForCacheSync(stopCh, c.clusterRoleBindingSynced, c.userSynced); !ok {
return fmt.Errorf("failed to wait for caches to sync")
}
@@ -195,7 +197,7 @@ func (c *Controller) processNextWorkItem() bool {
func (c *Controller) reconcile(key string) error {
// Get the clusterRoleBinding with this name
clusterRoleBinding, err := c.lister.Get(key)
clusterRoleBinding, err := c.clusterRoleBindingLister.Get(key)
if err != nil {
// The user may no longer exist, in which case we stop
// processing.

20
pkg/controller/globalrolebinding/globalrolebinding_controller.go
View File

@@ -25,7 +25,6 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
k8sinformers "k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -34,7 +33,6 @@ import (
"k8s.io/client-go/util/workqueue"
"k8s.io/klog"
iamv1alpha2 "kubesphere.io/kubesphere/pkg/apis/iam/v1alpha2"
ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
iamv1alpha2informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/iam/v1alpha2"
iamv1alpha2listers "kubesphere.io/kubesphere/pkg/client/listers/iam/v1alpha2"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -69,7 +67,7 @@ type Controller struct {
multiClusterEnabled bool
}
func NewController(k8sClient kubernetes.Interface, k8sInformer k8sinformers.SharedInformerFactory, ksInformer ksinformers.SharedInformerFactory, multiClusterEnabled bool) *Controller {
func NewController(k8sClient kubernetes.Interface, globalRoleBindingInformer iamv1alpha2informers.GlobalRoleBindingInformer, multiClusterEnabled bool) *Controller {
// Create event broadcaster
// Add sample-controller types to the default Kubernetes Scheme so Events can be
// logged for sample-controller types.
@@ -79,18 +77,17 @@ func NewController(k8sClient kubernetes.Interface, k8sInformer k8sinformers.Shar
eventBroadcaster.StartLogging(klog.Infof)
eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: k8sClient.CoreV1().Events("")})
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
informer := ksInformer.Iam().V1alpha2().GlobalRoleBindings()
ctl := &Controller{
k8sClient: k8sClient,
informer: informer,
lister: informer.Lister(),
synced: informer.Informer().HasSynced,
informer: globalRoleBindingInformer,
lister: globalRoleBindingInformer.Lister(),
synced: globalRoleBindingInformer.Informer().HasSynced,
workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "ClusterRoleBinding"),
recorder: recorder,
multiClusterEnabled: multiClusterEnabled,
}
klog.Info("Setting up event handlers")
informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
globalRoleBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: ctl.enqueueClusterRoleBinding,
UpdateFunc: func(old, new interface{}) {
ctl.enqueueClusterRoleBinding(new)
@@ -200,7 +197,6 @@ func (c *Controller) processNextWorkItem() bool {
// with the current status of the resource.
func (c *Controller) reconcile(key string) error {
// Get the clusterRoleBinding with this name
globalRoleBinding, err := c.lister.Get(key)
if err != nil {
// The user may no longer exist, in which case we stop
@@ -213,9 +209,7 @@ func (c *Controller) reconcile(key string) error {
return err
}
isPlatformAdmin := globalRoleBinding.RoleRef.Name == iamv1alpha2.PlatformAdmin
if isPlatformAdmin {
if globalRoleBinding.RoleRef.Name == iamv1alpha2.PlatformAdmin {
if err := c.relateToClusterAdmin(globalRoleBinding); err != nil {
klog.Error(err)
return err
@@ -257,8 +251,6 @@ func (c *Controller) relateToClusterAdmin(globalRoleBinding *iamv1alpha2.GlobalR
},
}
// rbac.authorization.k8s.io
err := controllerutil.SetControllerReference(globalRoleBinding, federatedClusterRoleBinding, scheme.Scheme)
if err != nil {

19
pkg/kapis/iam/v1alpha2/handler.go
View File

@@ -67,6 +67,11 @@ func (h *iamHandler) RetrieveMemberRoleTemplates(request *restful.Request, respo
globalRole, err := h.am.GetGlobalRoleOfUser(username)
if err != nil {
// if role binding not exist return empty list
if errors.IsNotFound(err) {
response.WriteEntity([]interface{}{})
return
}
api.HandleInternalError(response, request, err)
return
}
@@ -91,6 +96,11 @@ func (h *iamHandler) RetrieveMemberRoleTemplates(request *restful.Request, respo
clusterRole, err := h.am.GetClusterRoleOfUser(username)
if err != nil {
// if role binding not exist return empty list
if errors.IsNotFound(err) {
response.WriteEntity([]interface{}{})
return
}
api.HandleInternalError(response, request, err)
return
}
@@ -117,6 +127,11 @@ func (h *iamHandler) RetrieveMemberRoleTemplates(request *restful.Request, respo
workspaceRole, err := h.am.GetWorkspaceRoleOfUser(username, workspace)
if err != nil {
// if role binding not exist return empty list
if errors.IsNotFound(err) {
response.WriteEntity([]interface{}{})
return
}
api.HandleInternalError(response, request, err)
return
}
@@ -141,9 +156,9 @@ func (h *iamHandler) RetrieveMemberRoleTemplates(request *restful.Request, respo
namespace, err := h.resolveNamespace(request.PathParameter("namespace"), request.PathParameter("devops"))
if err != nil {
klog.Error(err)
// if role binding not exist return empty list
if errors.IsNotFound(err) {
api.HandleNotFound(response, request, err)
response.WriteEntity([]interface{}{})
return
}
api.HandleInternalError(response, request, err)

5
pkg/kapis/resources/v1alpha2/handler.go
View File

@@ -49,8 +49,9 @@ func newResourceHandler(k8sClient kubernetes.Interface, factory informers.Inform
gitVerifier: git.NewGitVerifier(factory.KubernetesSharedInformerFactory()),
registryGetter: registries.NewRegistryGetter(factory.KubernetesSharedInformerFactory()),
kubeconfigOperator: kubeconfig.NewOperator(k8sClient, nil, masterURL),
kubectlOperator: kubectl.NewOperator(k8sClient, factory.KubernetesSharedInformerFactory(),
factory.KubeSphereSharedInformerFactory()),
kubectlOperator: kubectl.NewOperator(nil, factory.KubernetesSharedInformerFactory().Apps().V1().Deployments(),
factory.KubernetesSharedInformerFactory().Core().V1().Pods(),
factory.KubeSphereSharedInformerFactory().Iam().V1alpha2().Users()),
}
}

6
pkg/models/iam/am/am.go
View File

@@ -334,8 +334,12 @@ func (am *amOperator) ListRoleBindings(username, namespace string) ([]*rbacv1.Ro
}
func contains(subjects []rbacv1.Subject, username string) bool {
// if username is nil means list all role bindings
if username == "" {
return true
}
for _, subject := range subjects {
if subject.Kind == rbacv1.UserKind && (username == "" || subject.Name == username) {
if subject.Kind == rbacv1.UserKind && subject.Name == username {
return true
}
}

22
pkg/models/kubectl/kubectl.go
View File

@@ -23,11 +23,12 @@ import (
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
k8sinformers "k8s.io/client-go/informers"
appsv1informers "k8s.io/client-go/informers/apps/v1"
coreinfomers "k8s.io/client-go/informers/core/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/klog"
"kubesphere.io/kubesphere/pkg/client/clientset/versioned/scheme"
ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
iamv1alpha2informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/iam/v1alpha2"
"kubesphere.io/kubesphere/pkg/models"
"math/rand"
"os"
@@ -47,13 +48,14 @@ type Interface interface {
}
type operator struct {
k8sClient kubernetes.Interface
k8sInformer k8sinformers.SharedInformerFactory
ksInformer ksinformers.SharedInformerFactory
k8sClient kubernetes.Interface
deploymentInformer appsv1informers.DeploymentInformer
podInformer coreinfomers.PodInformer
userInformer iamv1alpha2informers.UserInformer
}
func NewOperator(k8sClient kubernetes.Interface, k8sInformer k8sinformers.SharedInformerFactory, ksInformer ksinformers.SharedInformerFactory) Interface {
return &operator{k8sClient: k8sClient, k8sInformer: k8sInformer, ksInformer: ksInformer}
func NewOperator(k8sClient kubernetes.Interface, deploymentInformer appsv1informers.DeploymentInformer, podInformer coreinfomers.PodInformer, userInformer iamv1alpha2informers.UserInformer) Interface {
return &operator{k8sClient: k8sClient, deploymentInformer: deploymentInformer, podInformer: podInformer, userInformer: userInformer}
}
var DefaultImage = "kubesphere/kubectl:advanced-1.0.0"
@@ -66,7 +68,7 @@ func init() {
func (o *operator) GetKubectlPod(username string) (models.PodInfo, error) {
deployName := fmt.Sprintf(deployNameFormat, username)
deploy, err := o.k8sInformer.Apps().V1().Deployments().Lister().Deployments(namespace).Get(deployName)
deploy, err := o.deploymentInformer.Lister().Deployments(namespace).Get(deployName)
if err != nil {
klog.Errorln(err)
return models.PodInfo{}, err
@@ -74,7 +76,7 @@ func (o *operator) GetKubectlPod(username string) (models.PodInfo, error) {
selectors := deploy.Spec.Selector.MatchLabels
labelSelector := labels.Set(selectors).AsSelector()
pods, err := o.k8sInformer.Core().V1().Pods().Lister().Pods(namespace).List(labelSelector)
pods, err := o.podInformer.Lister().Pods(namespace).List(labelSelector)
if err != nil {
klog.Errorln(err)
return models.PodInfo{}, err
@@ -115,7 +117,7 @@ func selectCorrectPod(namespace string, pods []*v1.Pod) (kubectlPod *v1.Pod, err
func (o *operator) CreateKubectlDeploy(username string) error {
deployName := fmt.Sprintf(deployNameFormat, username)
user, err := o.ksInformer.Iam().V1alpha2().Users().Lister().Get(username)
user, err := o.userInformer.Lister().Get(username)
if err != nil {
klog.Error(err)