From 7bb760620f1948dd17d85f530232e49b7fdc4d0c Mon Sep 17 00:00:00 2001 From: Pengfei Ni Date: Fri, 7 Jul 2017 15:44:11 +0800 Subject: [PATCH] Rework auth controller This PR reworks auth controller to use informer framework. It also fix the problem of tenant and user not created. Change-Id: I017032f2eb4d83440319729d9f1fb13351f4d72b Closes-Bug: 1702841 Signed-off-by: Pengfei Ni --- .../stackube-controller.go | 16 +- pkg/apis/v1/types.go | 15 + pkg/auth-controller/client/auth/client.go | 58 +-- pkg/auth-controller/client/auth/tenant.go | 194 ++------- pkg/auth-controller/rbacmanager/controller.go | 28 +- pkg/auth-controller/tenant/controller.go | 370 ------------------ .../tenant/tenant_controller.go | 174 ++++++++ .../tenant/tenant_controller_helper.go | 83 ++++ 8 files changed, 335 insertions(+), 603 deletions(-) delete mode 100644 pkg/auth-controller/tenant/controller.go create mode 100644 pkg/auth-controller/tenant/tenant_controller.go create mode 100644 pkg/auth-controller/tenant/tenant_controller_helper.go diff --git a/cmd/stackube-controller/stackube-controller.go b/cmd/stackube-controller/stackube-controller.go index e539ea7..404baa4 100644 --- a/cmd/stackube-controller/stackube-controller.go +++ b/cmd/stackube-controller/stackube-controller.go @@ -26,15 +26,15 @@ var ( "path to stackube config file") ) -func startControllers(cfg tenant.Config) error { +func startControllers(kubeconfig, cloudconfig string) error { // Creates a new tenant controller - tc, err := tenant.New(cfg) + tc, err := tenant.New(kubeconfig, cloudconfig) if err != nil { return err } // Creates a new RBAC controller - rm, err := rbacmanager.New(cfg) + rm, err := rbacmanager.New(kubeconfig) if err != nil { return err } @@ -47,9 +47,7 @@ func startControllers(cfg tenant.Config) error { wg.Go(func() error { return rm.Run(ctx.Done()) }) networkController, err := network.NewNetworkController( - cfg.KubeConfig, - cfg.CloudConfig, - ) + kubeconfig, cloudconfig) if err != nil { return err } @@ -106,11 +104,7 @@ func main() { } // Start stackube controllers. - cfg := tenant.Config{ - KubeConfig: *kubeconfig, - CloudConfig: *cloudconfig, - } - if err := startControllers(cfg); err != nil { + if err := startControllers(*kubeconfig, *cloudconfig); err != nil { glog.Fatal(err) } } diff --git a/pkg/apis/v1/types.go b/pkg/apis/v1/types.go index 7d2ef59..fdf0cea 100644 --- a/pkg/apis/v1/types.go +++ b/pkg/apis/v1/types.go @@ -26,6 +26,21 @@ const ( NetworkTerminating = "Terminating" ) +// These are the valid phases of a tenant state. +const ( + // TenantInitializing means the tenant is just accepted by system + TenantInitializing = "Initializing" + // TenantActive means the tenant is available for use in the system + TenantActive = "Active" + // TenantPending means the tenant is accepted by system, but it is still + // processing by tenant provider + TenantPending = "Pending" + // TenantFailed means the tenant is not available + TenantFailed = "Failed" + // TenantTerminating means the tenant is undergoing graceful termination + TenantTerminating = "Terminating" +) + // Network describes a Neutron network. type Network struct { // TypeMeta defines type of the object and its API schema version. diff --git a/pkg/auth-controller/client/auth/client.go b/pkg/auth-controller/client/auth/client.go index f29149c..2b00fa1 100644 --- a/pkg/auth-controller/client/auth/client.go +++ b/pkg/auth-controller/client/auth/client.go @@ -1,61 +1,29 @@ package auth import ( - "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/serializer" - "k8s.io/client-go/dynamic" - "k8s.io/client-go/pkg/api" "k8s.io/client-go/rest" crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1" ) -var ( - CRDGroup = crv1.GroupName - CRDVersion = crv1.SchemeGroupVersion.Version -) +func NewClient(cfg *rest.Config) (*rest.RESTClient, *runtime.Scheme, error) { + scheme := runtime.NewScheme() + if err := crv1.AddToScheme(scheme); err != nil { + return nil, nil, err + } -type AuthInterface interface { - RESTClient() rest.Interface - TenantsGetter - //TODO: add networkgetter -} + config := *cfg + config.GroupVersion = &crv1.SchemeGroupVersion + config.APIPath = "/apis" + config.ContentType = runtime.ContentTypeJSON + config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: serializer.NewCodecFactory(scheme)} -type AuthClient struct { - restClient rest.Interface - dynamicClient *dynamic.Client -} - -func (c *AuthClient) Tenants(namespace string) TenantInterface { - return newTenants(c.restClient, c.dynamicClient, namespace) -} - -func (c *AuthClient) RESTClient() rest.Interface { - return c.restClient -} - -func NewForConfig(c *rest.Config) (*AuthClient, error) { - config := *c - setConfigDefaults(&config) client, err := rest.RESTClientFor(&config) if err != nil { - return nil, err + return nil, nil, err } - dynamicClient, err := dynamic.NewClient(&config) - if err != nil { - return nil, err - } - - return &AuthClient{client, dynamicClient}, nil -} - -func setConfigDefaults(config *rest.Config) { - config.GroupVersion = &schema.GroupVersion{ - Group: CRDGroup, - Version: CRDVersion, - } - config.APIPath = "/apis" - config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: api.Codecs} - return + return client, scheme, nil } diff --git a/pkg/auth-controller/client/auth/tenant.go b/pkg/auth-controller/client/auth/tenant.go index 579f73a..a13b264 100644 --- a/pkg/auth-controller/client/auth/tenant.go +++ b/pkg/auth-controller/client/auth/tenant.go @@ -1,175 +1,65 @@ package auth import ( - "encoding/json" + "reflect" + "time" - "git.openstack.org/openstack/stackube/pkg/apis/v1" + crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1" + "git.openstack.org/openstack/stackube/pkg/util" + apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/watch" - "k8s.io/client-go/dynamic" + "k8s.io/apimachinery/pkg/util/wait" + apiv1 "k8s.io/client-go/pkg/api/v1" "k8s.io/client-go/rest" ) const ( - CRDTenantsKind = "Tenant" - CRDTenantName = "tenants" + tenantCRDName = crv1.TenantResourcePlural + "." + crv1.GroupName ) -type TenantsGetter interface { - Tenants(namespace string) TenantInterface -} - -type TenantInterface interface { - Create(*v1.Tenant) (*v1.Tenant, error) - Get(name string) (*v1.Tenant, error) - Update(*v1.Tenant) (*v1.Tenant, error) - Delete(name string, options *metav1.DeleteOptions) error - List(opts metav1.ListOptions) (runtime.Object, error) - Watch(opts metav1.ListOptions) (watch.Interface, error) -} - -type tenants struct { - restClient rest.Interface - client *dynamic.ResourceClient - ns string -} - -func newTenants(r rest.Interface, c *dynamic.Client, namespace string) *tenants { - return &tenants{ - r, - c.Resource( - &metav1.APIResource{ - Kind: CRDTenantsKind, - Name: CRDTenantName, - Namespaced: true, +func CreateTenantCRD(clientset apiextensionsclient.Interface) (*apiextensionsv1beta1.CustomResourceDefinition, error) { + crd := &apiextensionsv1beta1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: tenantCRDName, + }, + Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ + Group: crv1.GroupName, + Version: crv1.SchemeGroupVersion.Version, + Scope: apiextensionsv1beta1.NamespaceScoped, + Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ + Plural: crv1.TenantResourcePlural, + Kind: reflect.TypeOf(crv1.Tenant{}).Name(), }, - namespace, - ), - namespace, + }, } -} - -func (p *tenants) Create(o *v1.Tenant) (*v1.Tenant, error) { - up, err := UnstructuredFromTenant(o) + _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) if err != nil { return nil, err } - up, err = p.client.Create(up) - if err != nil { + // wait for CRD being established + if err = util.WaitForCRDReady(clientset, tenantCRDName); err != nil { return nil, err + } else { + return crd, nil } - - return TenantFromUnstructured(up) } -func (p *tenants) Get(name string) (*v1.Tenant, error) { - obj, err := p.client.Get(name) - if err != nil { - return nil, err - } - return TenantFromUnstructured(obj) -} - -func (p *tenants) Update(o *v1.Tenant) (*v1.Tenant, error) { - up, err := UnstructuredFromTenant(o) - if err != nil { - return nil, err - } - - up, err = p.client.Update(up) - if err != nil { - return nil, err - } - - return TenantFromUnstructured(up) -} - -func (p *tenants) Delete(name string, options *metav1.DeleteOptions) error { - return p.client.Delete(name, options) -} - -func (p *tenants) List(opts metav1.ListOptions) (runtime.Object, error) { - req := p.restClient.Get(). - Namespace(p.ns). - Resource("tenants"). - // VersionedParams(&options, v1.ParameterCodec) - FieldsSelectorParam(nil) - - b, err := req.DoRaw() - if err != nil { - return nil, err - } - var tena v1.TenantList - return &tena, json.Unmarshal(b, &tena) -} - -func (p *tenants) Watch(opts metav1.ListOptions) (watch.Interface, error) { - r, err := p.restClient.Get(). - Prefix("watch"). - Namespace(p.ns). - Resource("tenants"). - // VersionedParams(&options, v1.ParameterCodec). - FieldsSelectorParam(nil). - Stream() - if err != nil { - return nil, err - } - return watch.NewStreamWatcher(&tenantDecoder{ - dec: json.NewDecoder(r), - close: r.Close, - }), nil -} - -// TenantFromUnstructured unmarshals a Tenant object from dynamic client's unstructured -func TenantFromUnstructured(r *unstructured.Unstructured) (*v1.Tenant, error) { - b, err := json.Marshal(r.Object) - if err != nil { - return nil, err - } - var p v1.Tenant - if err := json.Unmarshal(b, &p); err != nil { - return nil, err - } - p.TypeMeta.Kind = CRDTenantsKind - p.TypeMeta.APIVersion = CRDGroup + "/" + CRDVersion - return &p, nil -} - -// UnstructuredFromTenant marshals a Tenant object into dynamic client's unstructured -func UnstructuredFromTenant(p *v1.Tenant) (*unstructured.Unstructured, error) { - p.TypeMeta.Kind = CRDTenantsKind - p.TypeMeta.APIVersion = CRDGroup + "/" + CRDVersion - b, err := json.Marshal(p) - if err != nil { - return nil, err - } - var r unstructured.Unstructured - if err := json.Unmarshal(b, &r.Object); err != nil { - return nil, err - } - return &r, nil -} - -type tenantDecoder struct { - dec *json.Decoder - close func() error -} - -func (d *tenantDecoder) Close() { - d.close() -} - -func (d *tenantDecoder) Decode() (action watch.EventType, object runtime.Object, err error) { - var e struct { - Type watch.EventType - Object v1.Tenant - } - if err := d.dec.Decode(&e); err != nil { - return watch.Error, nil, err - } - return e.Type, &e.Object, nil +func WaitForTenantInstanceProcessed(kubeClient *rest.RESTClient, name string) error { + return wait.Poll(100*time.Millisecond, 10*time.Second, func() (bool, error) { + var tenant crv1.Tenant + err := kubeClient.Get(). + Resource(crv1.TenantResourcePlural). + Namespace(apiv1.NamespaceDefault). + Name(name). + Do().Into(&tenant) + + if err == nil && tenant.Status.State == crv1.TenantActive { + return true, nil + } + + return false, err + }) } diff --git a/pkg/auth-controller/rbacmanager/controller.go b/pkg/auth-controller/rbacmanager/controller.go index c1c3d08..a2eafb9 100644 --- a/pkg/auth-controller/rbacmanager/controller.go +++ b/pkg/auth-controller/rbacmanager/controller.go @@ -5,7 +5,6 @@ import ( "time" "git.openstack.org/openstack/stackube/pkg/auth-controller/rbacmanager/rbac" - "git.openstack.org/openstack/stackube/pkg/auth-controller/tenant" "git.openstack.org/openstack/stackube/pkg/util" "github.com/golang/glog" @@ -29,8 +28,8 @@ type Controller struct { } // New creates a new RBAC controller. -func New(conf tenant.Config) (*Controller, error) { - cfg, err := util.NewClusterConfig(conf.KubeConfig) +func New(kubeconfig string) (*Controller, error) { + cfg, err := util.NewClusterConfig(kubeconfig) if err != nil { return nil, fmt.Errorf("init cluster config failed: %v", err) } @@ -62,29 +61,8 @@ func New(conf tenant.Config) (*Controller, error) { func (c *Controller) Run(stopc <-chan struct{}) error { defer c.queue.ShutDown() - errChan := make(chan error) - go func() { - v, err := c.kclient.Discovery().ServerVersion() - if err != nil { - errChan <- fmt.Errorf("communicating with server failed: %v", err) - return - } - glog.V(4).Infof("Established connection established, cluster-version: %s", v) - errChan <- nil - }() - - select { - case err := <-errChan: - if err != nil { - return err - } - glog.V(4).Info("CRD API endpoints ready") - case <-stopc: - return nil - } - + glog.V(4).Info("Starting rbac manager") go c.worker() - go c.nsInf.Run(stopc) <-stopc diff --git a/pkg/auth-controller/tenant/controller.go b/pkg/auth-controller/tenant/controller.go deleted file mode 100644 index 42a4321..0000000 --- a/pkg/auth-controller/tenant/controller.go +++ /dev/null @@ -1,370 +0,0 @@ -package tenant - -import ( - "fmt" - "reflect" - "strings" - "time" - - "git.openstack.org/openstack/stackube/pkg/apis/v1" - crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1" - "git.openstack.org/openstack/stackube/pkg/auth-controller/client/auth" - "git.openstack.org/openstack/stackube/pkg/auth-controller/rbacmanager/rbac" - "git.openstack.org/openstack/stackube/pkg/openstack" - "git.openstack.org/openstack/stackube/pkg/util" - - "github.com/golang/glog" - apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" - apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" - apierrors "k8s.io/apimachinery/pkg/api/errors" - apismetav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - utilruntime "k8s.io/apimachinery/pkg/util/runtime" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/pkg/api" - apiv1 "k8s.io/client-go/pkg/api/v1" - "k8s.io/client-go/tools/cache" - "k8s.io/client-go/util/workqueue" -) - -var ( - // NOTE: we should always use crv1.TenantResourcePlural.CRDGroup as CRD name - crdTenant = crv1.TenantResourcePlural + "." + auth.CRDGroup - - resyncPeriod = 5 * time.Minute -) - -// TenantController manages lify cycle of Tenant. -type TenantController struct { - kclient *kubernetes.Clientset - crdclient *apiextensionsclient.Clientset - tclient *auth.AuthClient - osclient *openstack.Client - tenInf cache.SharedIndexInformer - queue workqueue.RateLimitingInterface - config Config -} - -// Config defines configuration parameters for the TenantController. -type Config struct { - KubeConfig string - CloudConfig string -} - -// New creates a new tenant controller. -func New(conf Config) (*TenantController, error) { - cfg, err := util.NewClusterConfig(conf.KubeConfig) - if err != nil { - return nil, fmt.Errorf("init cluster config failed: %v", err) - } - client, err := kubernetes.NewForConfig(cfg) - if err != nil { - return nil, fmt.Errorf("init kubernetes client failed: %v", err) - } - tclient, err := auth.NewForConfig(cfg) - if err != nil { - return nil, fmt.Errorf("init restclient for tenant failed: %v", err) - } - crdclient, err := apiextensionsclient.NewForConfig(cfg) - if err != nil { - return nil, fmt.Errorf("init CRD client failed: %v", err) - } - - openStackClient, err := openstack.NewClient(conf.CloudConfig) - if err != nil { - return nil, fmt.Errorf("init openstack client failed: %v", err) - } - - c := &TenantController{ - crdclient: crdclient, - kclient: client, - tclient: tclient, - osclient: openStackClient, - queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "tenant"), - config: conf, - } - - c.tenInf = cache.NewSharedIndexInformer( - &cache.ListWatch{ - ListFunc: tclient.Tenants(api.NamespaceAll).List, - WatchFunc: tclient.Tenants(api.NamespaceAll).Watch, - }, - &v1.Tenant{}, resyncPeriod, cache.Indexers{}, - ) - c.tenInf.AddEventHandler(cache.ResourceEventHandlerFuncs{ - AddFunc: c.handleAddTenant, - DeleteFunc: c.handleDeleteTenant, - UpdateFunc: c.handleUpdateTenant, - }) - - return c, nil -} - -// Run the controller. -func (c *TenantController) Run(stopc <-chan struct{}) error { - defer c.queue.ShutDown() - - errChan := make(chan error) - go func() { - v, err := c.kclient.Discovery().ServerVersion() - if err != nil { - errChan <- fmt.Errorf("communicating with server failed: %v", err) - return - } - glog.V(4).Infof("Established connection established, cluster-version: %s", v) - // Create CRD - if _, err := c.createTenantCRD(c.crdclient); err != nil { - if err != nil && !apierrors.IsAlreadyExists(err) { - errChan <- fmt.Errorf("creating tenant CRD failed: %v", err) - } - return - } - // Create clusterRole - if err = c.createClusterRoles(); err != nil { - errChan <- fmt.Errorf("creating clusterrole failed: %v", err) - return - } - - errChan <- nil - }() - - select { - case err := <-errChan: - if err != nil { - return err - } - glog.V(4).Info("CRD API endpoints ready") - case <-stopc: - return nil - } - - go c.worker() - - go c.tenInf.Run(stopc) - - <-stopc - return nil -} - -func (c *TenantController) keyFunc(obj interface{}) (string, bool) { - k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj) - if err != nil { - glog.V(4).Infof("Failed create key: %v", err) - return k, false - } - return k, true -} - -func (c *TenantController) handleAddTenant(obj interface{}) { - key, ok := c.keyFunc(obj) - if !ok { - return - } - glog.V(4).Infof("Added tenant %s", key) - c.enqueue(key) -} - -func (c *TenantController) handleDeleteTenant(obj interface{}) { - key, ok := c.keyFunc(obj) - if !ok { - return - } - glog.V(4).Infof("Deleted tenant %s", key) - c.enqueue(key) -} - -func (c *TenantController) handleUpdateTenant(old, cur interface{}) { - key, ok := c.keyFunc(cur) - if !ok { - return - } - glog.V(4).Infof("Updated tenant %s", key) - c.enqueue(key) -} - -// 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 *TenantController) 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 *TenantController) worker() { - for c.processNextWorkItem() { - } -} - -func (c *TenantController) 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 *TenantController) sync(key string) error { - obj, exists, err := c.tenInf.GetIndexer().GetByKey(key) - if err != nil { - return err - } - if !exists { - // Delete tenant related resources in k8s - tenant := strings.Split(key, "/") - deleteOptions := &apismetav1.DeleteOptions{ - TypeMeta: apismetav1.TypeMeta{ - Kind: "ClusterRoleBinding", - APIVersion: "rbac.authorization.k8s.io/v1beta1", - }, - } - err = c.kclient.Rbac().ClusterRoleBindings().Delete(tenant[1]+"-namespace-creater", deleteOptions) - if err != nil && !apierrors.IsNotFound(err) { - glog.Errorf("Failed delete ClusterRoleBinding for tenant %s: %v", tenant[1], err) - return err - } - glog.V(4).Infof("Deleted ClusterRoleBinding %s", tenant[1]) - //Delete namespace - err = c.deleteNamespace(tenant[1]) - if err != nil { - return err - } - glog.V(4).Infof("Deleted namespace %s", tenant[1]) - // Delete all users on a tenant - err = c.osclient.DeleteAllUsersOnTenant(tenant[1]) - if err != nil { - glog.Errorf("Failed delete all users in the tenant %s: %v", tenant[1], err) - return err - } - // Delete tenant in keystone - err = c.osclient.DeleteTenant(tenant[1]) - if err != nil { - glog.Errorf("Failed delete tenant %s: %v", tenant[1], err) - return err - } - return nil - } - - t := obj.(*v1.Tenant) - glog.V(4).Infof("Sync tenant %s", key) - err = c.syncTenant(t) - if err != nil { - return err - } - return nil -} - -func (c *TenantController) createTenantCRD(clientset apiextensionsclient.Interface) (*apiextensionsv1beta1.CustomResourceDefinition, error) { - crd := &apiextensionsv1beta1.CustomResourceDefinition{ - ObjectMeta: apismetav1.ObjectMeta{ - Name: crdTenant, - }, - Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ - Group: crv1.GroupName, - Version: crv1.SchemeGroupVersion.Version, - Scope: apiextensionsv1beta1.NamespaceScoped, - Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ - Plural: crv1.TenantResourcePlural, - Kind: reflect.TypeOf(crv1.Tenant{}).Name(), - }, - }, - } - _, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) - if err != nil { - return nil, err - } - - // wait for CRD being established - if err = util.WaitForCRDReady(clientset, crdTenant); err != nil { - return nil, err - } else { - return crd, nil - } -} - -func (c *TenantController) syncTenant(tenant *v1.Tenant) error { - roleBinding := rbac.GenerateClusterRoleBindingByTenant(tenant.Name) - _, err := c.kclient.Rbac().ClusterRoleBindings().Create(roleBinding) - if err != nil && !apierrors.IsAlreadyExists(err) { - glog.Errorf("Failed create ClusterRoleBinding for tenant %s: %v", tenant.Name, err) - return err - } - glog.V(4).Infof("Created ClusterRoleBindings %s-namespace-creater for tenant %s", tenant.Name, tenant.Name) - if tenant.Spec.TenantID != "" { - // Create user with the spec username and password in the given tenant - err = c.osclient.CreateUser(tenant.Spec.UserName, tenant.Spec.Password, tenant.Spec.TenantID) - if err != nil && !openstack.IsAlreadyExists(err) { - glog.Errorf("Failed create user %s: %v", tenant.Spec.UserName, err) - return err - } - } else { - // Create tenant if the tenant not exist in keystone - tenantID, err := c.osclient.CreateTenant(tenant.Name) - if err != nil { - return err - } - // Create user with the spec username and password in the created tenant - err = c.osclient.CreateUser(tenant.Spec.UserName, tenant.Spec.Password, tenantID) - if err != nil { - return err - } - } - - // Create namespace which name is the same as the tenant's name - err = c.createNamespace(tenant.Name) - if err != nil { - return err - } - glog.V(4).Infof("Created namespace %s for tenant %s", tenant.Name, tenant.Name) - return nil -} - -func (c *TenantController) createClusterRoles() error { - nsCreater := rbac.GenerateClusterRole() - _, err := c.kclient.Rbac().ClusterRoles().Create(nsCreater) - if err != nil && !apierrors.IsAlreadyExists(err) { - glog.Errorf("Failed create ClusterRoles namespace-creater: %v", err) - return err - } - glog.V(4).Info("Created ClusterRoles namespace-creater") - return nil -} - -func (c *TenantController) createNamespace(namespace string) error { - _, err := c.kclient.CoreV1().Namespaces().Create(&apiv1.Namespace{ - ObjectMeta: apismetav1.ObjectMeta{ - Name: namespace, - }, - }) - if err != nil && !apierrors.IsAlreadyExists(err) { - glog.Errorf("Failed create namespace %s: %v", namespace, err) - return err - } - return nil -} - -func (c *TenantController) deleteNamespace(namespace string) error { - err := c.kclient.CoreV1().Namespaces().Delete(namespace, apismetav1.NewDeleteOptions(0)) - if err != nil { - glog.Errorf("Failed delete namespace %s: %v", namespace, err) - return err - } - return nil -} diff --git a/pkg/auth-controller/tenant/tenant_controller.go b/pkg/auth-controller/tenant/tenant_controller.go new file mode 100644 index 0000000..46902b3 --- /dev/null +++ b/pkg/auth-controller/tenant/tenant_controller.go @@ -0,0 +1,174 @@ +package tenant + +import ( + "fmt" + + crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1" + crdClient "git.openstack.org/openstack/stackube/pkg/auth-controller/client/auth" + "git.openstack.org/openstack/stackube/pkg/openstack" + + "github.com/golang/glog" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apismetav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/client-go/kubernetes" + apiv1 "k8s.io/client-go/pkg/api/v1" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/clientcmd" +) + +// TenantController manages lify cycle of Tenant. +type TenantController struct { + k8sClient *kubernetes.Clientset + tenantClient *rest.RESTClient + tenantScheme *runtime.Scheme + openstackClient *openstack.Client +} + +// New creates a new tenant controller. +func New(kubeconfig, cloudconfig string) (*TenantController, error) { + // Create OpenStack client from config + openStackClient, err := openstack.NewClient(cloudconfig) + if err != nil { + return nil, fmt.Errorf("init openstack client failed: %v", err) + } + + // Create the client config. Use kubeconfig if given, otherwise assume in-cluster. + config, err := buildConfig(kubeconfig) + if err != nil { + return nil, fmt.Errorf("failed to build kubeconfig: %v", err) + } + clientset, err := apiextensionsclient.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("failed to create kubeclient from config: %v", err) + } + + // initialize CRD if it does not exist + _, err = crdClient.CreateTenantCRD(clientset) + if err != nil && !apierrors.IsAlreadyExists(err) { + return nil, fmt.Errorf("failed to create CRD to kube-apiserver: %v", err) + } + + k8sClient, err := kubernetes.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("failed to create kubernetes client: %v", err) + } + + // make a new config for our extension's API group, using the first config as a baseline + tenantClient, tenantScheme, err := crdClient.NewClient(config) + if err != nil { + return nil, fmt.Errorf("failed to create client for CRD: %v", err) + } + + c := &TenantController{ + tenantClient: tenantClient, + tenantScheme: tenantScheme, + k8sClient: k8sClient, + openstackClient: openStackClient, + } + + if err = c.createClusterRoles(); err != nil { + return nil, fmt.Errorf("failed to create cluster roles to kube-apiserver: %v", err) + } + + return c, nil +} + +func buildConfig(kubeconfig string) (*rest.Config, error) { + if kubeconfig != "" { + return clientcmd.BuildConfigFromFlags("", kubeconfig) + } + return rest.InClusterConfig() +} + +// Run the controller. +func (c *TenantController) Run(stopCh <-chan struct{}) error { + defer utilruntime.HandleCrash() + + source := cache.NewListWatchFromClient( + c.tenantClient, + crv1.TenantResourcePlural, + apiv1.NamespaceAll, + fields.Everything()) + + _, tenantInformor := cache.NewInformer( + source, + &crv1.Tenant{}, + 0, + cache.ResourceEventHandlerFuncs{ + AddFunc: c.onAdd, + UpdateFunc: c.onUpdate, + DeleteFunc: c.onDelete, + }) + + go tenantInformor.Run(stopCh) + <-stopCh + return nil +} + +func (c *TenantController) onAdd(obj interface{}) { + tenant := obj.(*crv1.Tenant) + glog.V(3).Infof("Tenant controller received new object %q\n", tenant) + + copyObj, err := c.tenantScheme.Copy(tenant) + if err != nil { + glog.Errorf("ERROR creating a deep copy of tenant object: %v\n", err) + return + } + + newTenant := copyObj.(*crv1.Tenant) + c.syncTenant(newTenant) +} + +func (c *TenantController) onUpdate(obj1, obj2 interface{}) { + glog.Warning("tenant updates is not supported yet.") +} + +func (c *TenantController) onDelete(obj interface{}) { + tenant, ok := obj.(*crv1.Tenant) + if !ok { + return + } + + glog.V(3).Infof("Tenant controller received deleted tenant %q\n", tenant) + + deleteOptions := &apismetav1.DeleteOptions{ + TypeMeta: apismetav1.TypeMeta{ + Kind: "ClusterRoleBinding", + APIVersion: "rbac.authorization.k8s.io/v1beta1", + }, + } + tenantName := tenant.Name + err := c.k8sClient.Rbac().ClusterRoleBindings().Delete(tenantName+"-namespace-creater", deleteOptions) + if err != nil && !apierrors.IsNotFound(err) { + glog.Errorf("Failed delete ClusterRoleBinding for tenant %s: %v", tenantName, err) + } else { + glog.V(4).Infof("Deleted ClusterRoleBinding %s", tenantName) + } + + //Delete namespace + err = c.deleteNamespace(tenantName) + if err != nil { + glog.Errorf("Delete namespace %s failed: %v", tenantName, err) + } else { + glog.V(4).Infof("Deleted namespace %s", tenantName) + } + + // Delete all users on a tenant + err = c.openstackClient.DeleteAllUsersOnTenant(tenantName) + if err != nil { + glog.Errorf("Failed delete all users in the tenant %s: %v", tenantName, err) + } + + // Delete tenant in keystone + if tenant.Spec.TenantID == "" { + err = c.openstackClient.DeleteTenant(tenantName) + if err != nil { + glog.Errorf("Failed delete tenant %s: %v", tenantName, err) + } + } +} diff --git a/pkg/auth-controller/tenant/tenant_controller_helper.go b/pkg/auth-controller/tenant/tenant_controller_helper.go new file mode 100644 index 0000000..2f0af10 --- /dev/null +++ b/pkg/auth-controller/tenant/tenant_controller_helper.go @@ -0,0 +1,83 @@ +package tenant + +import ( + crv1 "git.openstack.org/openstack/stackube/pkg/apis/v1" + "git.openstack.org/openstack/stackube/pkg/auth-controller/rbacmanager/rbac" + "git.openstack.org/openstack/stackube/pkg/openstack" + + "github.com/golang/glog" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apismetav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + apiv1 "k8s.io/client-go/pkg/api/v1" +) + +func (c *TenantController) syncTenant(tenant *crv1.Tenant) error { + roleBinding := rbac.GenerateClusterRoleBindingByTenant(tenant.Name) + _, err := c.k8sClient.Rbac().ClusterRoleBindings().Create(roleBinding) + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Errorf("Failed create ClusterRoleBinding for tenant %s: %v", tenant.Name, err) + return err + } + glog.V(4).Infof("Created ClusterRoleBindings %s-namespace-creater for tenant %s", tenant.Name, tenant.Name) + if tenant.Spec.TenantID != "" { + // Create user with the spec username and password in the given tenant + err = c.openstackClient.CreateUser(tenant.Spec.UserName, tenant.Spec.Password, tenant.Spec.TenantID) + if err != nil && !openstack.IsAlreadyExists(err) { + glog.Errorf("Failed create user %s: %v", tenant.Spec.UserName, err) + return err + } + } else { + // Create tenant if the tenant not exist in keystone + tenantID, err := c.openstackClient.CreateTenant(tenant.Name) + if err != nil { + return err + } + // Create user with the spec username and password in the created tenant + err = c.openstackClient.CreateUser(tenant.Spec.UserName, tenant.Spec.Password, tenantID) + if err != nil { + return err + } + } + + // Create namespace which name is the same as the tenant's name + err = c.createNamespace(tenant.Name) + if err != nil { + return err + } + + glog.V(4).Infof("Created namespace %s for tenant %s", tenant.Name, tenant.Name) + return nil +} + +func (c *TenantController) createClusterRoles() error { + nsCreater := rbac.GenerateClusterRole() + _, err := c.k8sClient.Rbac().ClusterRoles().Create(nsCreater) + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Errorf("Failed create ClusterRoles namespace-creater: %v", err) + return err + } + glog.V(4).Info("Created ClusterRoles namespace-creater") + return nil +} + +func (c *TenantController) createNamespace(namespace string) error { + _, err := c.k8sClient.CoreV1().Namespaces().Create(&apiv1.Namespace{ + ObjectMeta: apismetav1.ObjectMeta{ + Name: namespace, + }, + }) + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Errorf("Failed create namespace %s: %v", namespace, err) + return err + } + return nil +} + +func (c *TenantController) deleteNamespace(namespace string) error { + err := c.k8sClient.CoreV1().Namespaces().Delete(namespace, apismetav1.NewDeleteOptions(0)) + if err != nil { + glog.Errorf("Failed delete namespace %s: %v", namespace, err) + return err + } + return nil +}