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 <feiskyer@gmail.com>
This commit is contained in:
Pengfei Ni
parent
f52aae3420
commit
7bb760620f
@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
}
|
||||
174
pkg/auth-controller/tenant/tenant_controller.go
Normal file
174
pkg/auth-controller/tenant/tenant_controller.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
83
pkg/auth-controller/tenant/tenant_controller_helper.go
Normal file
83
pkg/auth-controller/tenant/tenant_controller_helper.go
Normal file
@ -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
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user