mirror of
https://github.com/rancher/norman.git
synced 2025-08-02 07:59:20 +00:00
Move CRD initialization out of CRDStore
This commit is contained in:
Darren Shepherd
parent
7c31eb8d7b
commit
13be339349
@ -2,27 +2,20 @@ package crd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/rancher/norman/store/proxy"
|
||||
"github.com/rancher/norman/types"
|
||||
"github.com/rancher/norman/types/convert"
|
||||
"github.com/sirupsen/logrus"
|
||||
apiext "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
|
||||
"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
|
||||
apiextclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
"k8s.io/client-go/dynamic"
|
||||
"k8s.io/client-go/rest"
|
||||
)
|
||||
|
||||
type Store struct {
|
||||
apiExtClientSet apiextclientset.Interface
|
||||
k8sClient rest.Interface
|
||||
schemaStores map[string]*proxy.Store
|
||||
Factory *Factory
|
||||
k8sClient rest.Interface
|
||||
schemaStores map[string]*proxy.Store
|
||||
}
|
||||
|
||||
func NewCRDStoreFromConfig(config rest.Config) (*Store, error) {
|
||||
@ -47,9 +40,11 @@ func NewCRDStoreFromConfig(config rest.Config) (*Store, error) {
|
||||
|
||||
func NewCRDStoreFromClients(apiExtClientSet apiextclientset.Interface, k8sClient rest.Interface) *Store {
|
||||
return &Store{
|
||||
apiExtClientSet: apiExtClientSet,
|
||||
k8sClient: k8sClient,
|
||||
schemaStores: map[string]*proxy.Store{},
|
||||
Factory: &Factory{
|
||||
APIExtClientSet: apiExtClientSet,
|
||||
},
|
||||
k8sClient: k8sClient,
|
||||
schemaStores: map[string]*proxy.Store{},
|
||||
}
|
||||
}
|
||||
|
||||
@ -118,34 +113,11 @@ func (c *Store) AddSchemas(ctx context.Context, schemas ...*types.Schema) error
|
||||
allSchemas = append(allSchemas, schema)
|
||||
}
|
||||
|
||||
ready, err := c.getReadyCRDs()
|
||||
schemaStatus, err := c.Factory.AddSchemas(ctx, allSchemas...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, schema := range allSchemas {
|
||||
crd, err := c.createCRD(schema, ready)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
schemaStatus[schema] = crd
|
||||
}
|
||||
|
||||
ready, err = c.getReadyCRDs()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for schema, crd := range schemaStatus {
|
||||
if readyCrd, ok := ready[crd.Name]; ok {
|
||||
schemaStatus[schema] = readyCrd
|
||||
} else {
|
||||
if err := c.waitCRD(ctx, crd.Name, schema, schemaStatus); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for schema, crd := range schemaStatus {
|
||||
c.schemaStores[key(schema)] = proxy.NewProxyStore(c.k8sClient,
|
||||
[]string{"apis"},
|
||||
@ -157,106 +129,3 @@ func (c *Store) AddSchemas(ctx context.Context, schemas ...*types.Schema) error
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func contains(list []string, s string) bool {
|
||||
for _, i := range list {
|
||||
if i == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (c *Store) waitCRD(ctx context.Context, crdName string, schema *types.Schema, schemaStatus map[*types.Schema]*apiext.CustomResourceDefinition) error {
|
||||
logrus.Infof("Waiting for CRD %s to become available", crdName)
|
||||
defer logrus.Infof("Done waiting for CRD %s to become available", crdName)
|
||||
|
||||
first := true
|
||||
return wait.Poll(500*time.Millisecond, 60*time.Second, func() (bool, error) {
|
||||
if !first {
|
||||
logrus.Infof("Waiting for CRD %s to become available", crdName)
|
||||
}
|
||||
first = false
|
||||
|
||||
crd, err := c.apiExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().Get(crdName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
for _, cond := range crd.Status.Conditions {
|
||||
switch cond.Type {
|
||||
case apiext.Established:
|
||||
if cond.Status == apiext.ConditionTrue {
|
||||
schemaStatus[schema] = crd
|
||||
return true, err
|
||||
}
|
||||
case apiext.NamesAccepted:
|
||||
if cond.Status == apiext.ConditionFalse {
|
||||
logrus.Infof("Name conflict on %s: %v\n", crdName, cond.Reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false, ctx.Err()
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Store) createCRD(schema *types.Schema, ready map[string]*apiext.CustomResourceDefinition) (*apiext.CustomResourceDefinition, error) {
|
||||
plural := strings.ToLower(schema.PluralName)
|
||||
name := strings.ToLower(plural + "." + schema.Version.Group)
|
||||
|
||||
crd, ok := ready[name]
|
||||
if ok {
|
||||
return crd, nil
|
||||
}
|
||||
|
||||
crd = &apiext.CustomResourceDefinition{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: name,
|
||||
},
|
||||
Spec: apiext.CustomResourceDefinitionSpec{
|
||||
Group: schema.Version.Group,
|
||||
Version: schema.Version.Version,
|
||||
Names: apiext.CustomResourceDefinitionNames{
|
||||
Plural: plural,
|
||||
Kind: convert.Capitalize(schema.ID),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
if schema.Scope == types.NamespaceScope {
|
||||
crd.Spec.Scope = apiext.NamespaceScoped
|
||||
} else {
|
||||
crd.Spec.Scope = apiext.ClusterScoped
|
||||
}
|
||||
|
||||
logrus.Infof("Creating CRD %s", name)
|
||||
crd, err := c.apiExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd)
|
||||
if errors.IsAlreadyExists(err) {
|
||||
return crd, nil
|
||||
}
|
||||
return crd, err
|
||||
}
|
||||
|
||||
func (c *Store) getReadyCRDs() (map[string]*apiext.CustomResourceDefinition, error) {
|
||||
list, err := c.apiExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().List(metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result := map[string]*apiext.CustomResourceDefinition{}
|
||||
|
||||
for i, crd := range list.Items {
|
||||
for _, cond := range crd.Status.Conditions {
|
||||
switch cond.Type {
|
||||
case apiext.Established:
|
||||
if cond.Status == apiext.ConditionTrue {
|
||||
result[crd.Name] = &list.Items[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
147
store/crd/init.go
Normal file
147
store/crd/init.go
Normal file
@ -0,0 +1,147 @@
|
||||
package crd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/rancher/norman/types"
|
||||
"github.com/rancher/norman/types/convert"
|
||||
"github.com/sirupsen/logrus"
|
||||
apiext "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
|
||||
apiextclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/wait"
|
||||
)
|
||||
|
||||
type Factory struct {
|
||||
APIExtClientSet apiextclientset.Interface
|
||||
}
|
||||
|
||||
func (c *Factory) AddSchemas(ctx context.Context, schemas ...*types.Schema) (map[*types.Schema]*apiext.CustomResourceDefinition, error) {
|
||||
schemaStatus := map[*types.Schema]*apiext.CustomResourceDefinition{}
|
||||
|
||||
ready, err := c.getReadyCRDs()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, schema := range schemas {
|
||||
crd, err := c.createCRD(schema, ready)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
schemaStatus[schema] = crd
|
||||
}
|
||||
|
||||
ready, err = c.getReadyCRDs()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for schema, crd := range schemaStatus {
|
||||
if readyCrd, ok := ready[crd.Name]; ok {
|
||||
schemaStatus[schema] = readyCrd
|
||||
} else {
|
||||
if err := c.waitCRD(ctx, crd.Name, schema, schemaStatus); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return schemaStatus, nil
|
||||
}
|
||||
|
||||
func (c *Factory) waitCRD(ctx context.Context, crdName string, schema *types.Schema, schemaStatus map[*types.Schema]*apiext.CustomResourceDefinition) error {
|
||||
logrus.Infof("Waiting for CRD %s to become available", crdName)
|
||||
defer logrus.Infof("Done waiting for CRD %s to become available", crdName)
|
||||
|
||||
first := true
|
||||
return wait.Poll(500*time.Millisecond, 60*time.Second, func() (bool, error) {
|
||||
if !first {
|
||||
logrus.Infof("Waiting for CRD %s to become available", crdName)
|
||||
}
|
||||
first = false
|
||||
|
||||
crd, err := c.APIExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().Get(crdName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
for _, cond := range crd.Status.Conditions {
|
||||
switch cond.Type {
|
||||
case apiext.Established:
|
||||
if cond.Status == apiext.ConditionTrue {
|
||||
schemaStatus[schema] = crd
|
||||
return true, err
|
||||
}
|
||||
case apiext.NamesAccepted:
|
||||
if cond.Status == apiext.ConditionFalse {
|
||||
logrus.Infof("Name conflict on %s: %v\n", crdName, cond.Reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false, ctx.Err()
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Factory) createCRD(schema *types.Schema, ready map[string]*apiext.CustomResourceDefinition) (*apiext.CustomResourceDefinition, error) {
|
||||
plural := strings.ToLower(schema.PluralName)
|
||||
name := strings.ToLower(plural + "." + schema.Version.Group)
|
||||
|
||||
crd, ok := ready[name]
|
||||
if ok {
|
||||
return crd, nil
|
||||
}
|
||||
|
||||
crd = &apiext.CustomResourceDefinition{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: name,
|
||||
},
|
||||
Spec: apiext.CustomResourceDefinitionSpec{
|
||||
Group: schema.Version.Group,
|
||||
Version: schema.Version.Version,
|
||||
Names: apiext.CustomResourceDefinitionNames{
|
||||
Plural: plural,
|
||||
Kind: convert.Capitalize(schema.ID),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
if schema.Scope == types.NamespaceScope {
|
||||
crd.Spec.Scope = apiext.NamespaceScoped
|
||||
} else {
|
||||
crd.Spec.Scope = apiext.ClusterScoped
|
||||
}
|
||||
|
||||
logrus.Infof("Creating CRD %s", name)
|
||||
crd, err := c.APIExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd)
|
||||
if errors.IsAlreadyExists(err) {
|
||||
return crd, nil
|
||||
}
|
||||
return crd, err
|
||||
}
|
||||
|
||||
func (c *Factory) getReadyCRDs() (map[string]*apiext.CustomResourceDefinition, error) {
|
||||
list, err := c.APIExtClientSet.ApiextensionsV1beta1().CustomResourceDefinitions().List(metav1.ListOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
result := map[string]*apiext.CustomResourceDefinition{}
|
||||
|
||||
for i, crd := range list.Items {
|
||||
for _, cond := range crd.Status.Conditions {
|
||||
switch cond.Type {
|
||||
case apiext.Established:
|
||||
if cond.Status == apiext.ConditionTrue {
|
||||
result[crd.Name] = &list.Items[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user