x/stackube
Code Issues Proposed changes

Rework RBAC controller

Browse Source 
This PR reworks RBAC controller to use informer framework.
It also update openstack NewClient method.

Change-Id: I6096a669b51f2cdacb7e492e4d3937f15b323b3c
Signed-off-by: mozhuli <21621232@zju.edu.cn>
This commit is contained in:
mozhulee 2017-07-30 16:51:34 +08:00
parent 0e29323a3e
commit f6d5dccb19
5 changed files with 55 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

188
pkg/auth-controller/rbacmanager/rbac_controller.go
View File

@ -17,7 +17,6 @@ limitations under the License.
package rbacmanager
import (
"fmt"
"time"
crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1"
@ -26,139 +25,85 @@ import (
"git.openstack.org/openstack/stackube/pkg/util"
"github.com/golang/glog"
"k8s.io/api/core/v1"
apiv1 "k8s.io/api/core/v1"
apierrors "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"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/util/workqueue"
)
const (
resyncPeriod = 5 * time.Minute
)
// Controller manages life cycle of namespace's rbac.
type Controller struct {
kclient *kubernetes.Clientset
nsInf cache.SharedIndexInformer
queue workqueue.RateLimitingInterface
k8sclient *kubernetes.Clientset
kubeCRDClient *crdClient.CRDClient
userCIDR string
userGateway string
}
// New creates a new RBAC controller.
// NewRBACController creates a new RBAC controller.
func NewRBACController(kubeClient *kubernetes.Clientset, kubeCRDClient *crdClient.CRDClient, userCIDR string,
userGateway string) (*Controller, error) {
o := &Controller{
kclient: kubeClient,
queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "rbacmanager"),
c := &Controller{
k8sclient: kubeClient,
kubeCRDClient: kubeCRDClient,
userCIDR: userCIDR,
userGateway: userGateway,
}
o.nsInf = cache.NewSharedIndexInformer(
cache.NewListWatchFromClient(o.kclient.Core().RESTClient(), "namespaces", v1.NamespaceAll, fields.Everything()),
&v1.Namespace{}, resyncPeriod, cache.Indexers{},
)
o.nsInf.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: o.handleNamespaceAdd,
DeleteFunc: o.handleNamespaceDelete,
UpdateFunc: o.handleNamespaceUpdate,
})
return o, nil
return c, nil
}
// Run the controller.
func (c *Controller) Run(stopc <-chan struct{}) error {
defer c.queue.ShutDown()
func (c *Controller) Run(stopCh <-chan struct{}) error {
defer utilruntime.HandleCrash()
glog.V(4).Info("Starting rbac manager")
go c.worker()
go c.nsInf.Run(stopc)
source := cache.NewListWatchFromClient(
c.k8sclient.Core().RESTClient(),
"namespaces",
apiv1.NamespaceAll,
fields.Everything())
<-stopc
_, namespaceInformor := cache.NewInformer(
source,
&apiv1.Namespace{},
resyncPeriod,
cache.ResourceEventHandlerFuncs{
AddFunc: c.onAdd,
UpdateFunc: c.onUpdate,
DeleteFunc: c.onDelete,
})
go namespaceInformor.Run(stopCh)
<-stopCh
return nil
}
func (c *Controller) keyFunc(obj interface{}) (string, bool) {
k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
if err != nil {
glog.Errorf("Creating key failed: %v", err)
return k, false
}
return k, true
}
func (c *Controller) onAdd(obj interface{}) {
namespace := obj.(*apiv1.Namespace)
glog.V(3).Infof("RBAC controller received new object %#v\n", namespace)
// enqueue adds a key to the queue. If obj is a key already it gets added directly.
// Otherwise, the key is extracted via keyFunc.
func (c *Controller) enqueue(obj interface{}) {
if obj == nil {
return
}
key, ok := obj.(string)
if !ok {
key, ok = c.keyFunc(obj)
if !ok {
return
}
}
c.queue.Add(key)
}
// worker runs a worker thread that just dequeues items, processes them, and marks them done.
// It enforces that the syncHandler is never invoked concurrently with the same key.
func (c *Controller) worker() {
for c.processNextWorkItem() {
}
}
func (c *Controller) processNextWorkItem() bool {
key, quit := c.queue.Get()
if quit {
return false
}
defer c.queue.Done(key)
err := c.sync(key.(string))
if err == nil {
c.queue.Forget(key)
return true
}
utilruntime.HandleError(fmt.Errorf("Sync %q failed: %v", key, err))
c.queue.AddRateLimited(key)
return true
}
func (c *Controller) handleNamespaceAdd(obj interface{}) {
key, ok := c.keyFunc(obj)
if !ok {
return
}
// check if this is a system reserved namespace
if util.IsSystemNamespace(key) {
// Check if this is a system reserved namespace
if util.IsSystemNamespace(namespace.Name) {
if err := c.initSystemReservedTenantNetwork(); err != nil {
glog.Error(err)
return
}
return
} else {
if err := c.createNetworkForTenant(key); err != nil {
if err := c.createNetworkForTenant(namespace.Name); err != nil {
glog.Error(err)
return
}
}
glog.V(4).Infof("Added namespace %s", namespace.Name)
glog.V(4).Infof("Added namespace %s", key)
c.enqueue(key)
c.syncRBAC(namespace)
}
// createNetworkForTenant automatically create network for given non-system tenant
@ -223,73 +168,36 @@ func (c *Controller) initSystemReservedTenantNetwork() error {
return nil
}
func (c *Controller) handleNamespaceDelete(obj interface{}) {
key, ok := c.keyFunc(obj)
if !ok {
return
}
glog.V(4).Infof("Deleted namespace %s", key)
c.enqueue(key)
func (c *Controller) onUpdate(obj1, obj2 interface{}) {
// NOTE(mozhuli) not supported yet
}
func (c *Controller) handleNamespaceUpdate(old, cur interface{}) {
oldns := old.(*v1.Namespace)
curns := cur.(*v1.Namespace)
if oldns.ResourceVersion == curns.ResourceVersion {
return
}
key, ok := c.keyFunc(cur)
if !ok {
return
}
glog.V(4).Infof("Updated namespace %s", key)
c.enqueue(key)
func (c *Controller) onDelete(obj interface{}) {
namespace := obj.(*apiv1.Namespace)
// tenant controller have done all the works so we will not wait here
glog.V(3).Infof("RBAC controller received deleted namespace %#v\n", namespace)
}
func (c *Controller) sync(key string) error {
obj, exists, err := c.nsInf.GetIndexer().GetByKey(key)
if err != nil {
return err
}
if !exists {
return nil
}
ns := obj.(*v1.Namespace)
glog.V(4).Infof("Sync RBAC %s", key)
err = c.syncRbac(ns)
if err != nil {
return err
}
return nil
}
func (c *Controller) syncRbac(ns *v1.Namespace) error {
func (c *Controller) syncRBAC(ns *apiv1.Namespace) error {
if ns.DeletionTimestamp != nil {
return nil
}
tenant, ok := ns.Labels["tenant"]
if !ok {
return nil
}
rbacClient := c.kclient.Rbac()
rbacClient := c.k8sclient.Rbac()
// Create role for tenant
role := rbac.GenerateRoleByNamespace(ns.Name)
_, err := rbacClient.Roles(ns.Name).Create(role)
if err != nil && !apierrors.IsAlreadyExists(err) {
glog.Errorf("Failed create default-role in namespace %s for tenant %s: %v", ns.Name, tenant, err)
glog.Errorf("Failed create default-role in namespace %s for tenant %s: %v", ns.Name, ns.Name, err)
return err
}
glog.V(4).Infof("Created default-role in namespace %s for tenant %s", ns.Name, tenant)
glog.V(4).Infof("Created default-role in namespace %s for tenant %s", ns.Name, ns.Name)
// Create rolebinding for tenant
roleBinding := rbac.GenerateRoleBinding(ns.Name, tenant)
roleBinding := rbac.GenerateRoleBinding(ns.Name, ns.Name)
_, err = rbacClient.RoleBindings(ns.Name).Create(roleBinding)
if err != nil && !apierrors.IsAlreadyExists(err) {
glog.Errorf("Failed create %s-rolebindings in namespace %s for tenant %s: %v", tenant, ns.Name, tenant, err)
glog.Errorf("Failed create %s-rolebindings in namespace %s for tenant %s: %v", ns.Name, ns.Name, ns.Name, err)
return err
}
glog.V(4).Infof("Created %s-rolebindings in namespace %s for tenant %s", tenant, ns.Name, tenant)
glog.V(4).Infof("Created %s-rolebindings in namespace %s for tenant %s", ns.Name, ns.Name, ns.Name)
return nil
}

2
pkg/auth-controller/tenant/tenant_controller.go
View File

@ -136,7 +136,7 @@ func (c *TenantController) onDelete(obj interface{}) {
// Delete automatically created network
// TODO(harry) so that we can not deal with network with different name and namespace,
// we need to document that.
if err := c.kubeCRDClient.DeleteNetork(tenantName); err != nil {
if err := c.kubeCRDClient.DeleteNetwork(tenantName); err != nil {
glog.Errorf("failed to delete network for tenant: %v", tenantName)
}

2
pkg/kubecrd/crdclient.go
View File

@ -134,7 +134,7 @@ func (c *CRDClient) AddNetwork(network *crv1.Network) error {
return nil
}
func (c *CRDClient) DeleteNetork(namespace string) error {
func (c *CRDClient) DeleteNetwork(namespace string) error {
// NOTE: the automatically created network for tenant use namespace as name
err := c.Client.Delete().
Resource(crv1.NetworkResourcePlural).

4
pkg/network-controller/network_controller_helper.go
View File

@ -44,12 +44,12 @@ func (c *NetworkController) addNetworkToDriver(kubeNetwork *crv1.Network) {
tenantID, err = c.driver.GetTenantIDFromName(kubeNetwork.GetNamespace())
if err != nil {
glog.Errorf("failed to fetch tenantID for tenantName: %v, error: %v retrying\n", tenantName, err)
return false, nil
return false, err
}
if tenantID == "" {
glog.V(5).Infof("tenantID is empty for tenantName: %v, retrying\n", tenantName)
return false, nil
return false, err
}
return true, nil
})

11
pkg/openstack/client.go
View File

@ -97,6 +97,7 @@ func toAuthOptions(cfg Config) gophercloud.AuthOptions {
Username: cfg.Global.Username,
Password: cfg.Global.Password,
TenantName: cfg.Global.TenantName,
AllowReauth: true,
}
}
@ -104,21 +105,15 @@ func NewClient(config string, kubeConfig string) (*Client, error) {
var opts gophercloud.AuthOptions
cfg, err := readConfig(config)
if err != nil {
glog.V(0).Infof("Failed read cloudconfig: %v. Starting init openstackclient from env", err)
opts, err = openstack.AuthOptionsFromEnv()
if err != nil {
return nil, err
}
} else {
opts = toAuthOptions(cfg)
return nil, fmt.Errorf("Failed read cloudconfig: %v", err)
}
glog.V(1).Infof("Initializing openstack client with config %v", cfg)
if cfg.Global.ExtNetID == "" {
return nil, fmt.Errorf("external network ID not set")
}
opts = toAuthOptions(cfg)
provider, err := openstack.AuthenticatedClient(opts)
if err != nil {
return nil, err