mirror of
https://github.com/rancher/steve.git
synced 2025-09-17 15:58:41 +00:00
323 lines
8.7 KiB
Go
323 lines
8.7 KiB
Go
package schema
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/rancher/apiserver/pkg/types"
|
|
"github.com/rancher/steve/pkg/attributes"
|
|
"github.com/rancher/steve/pkg/resources/common"
|
|
schema2 "github.com/rancher/steve/pkg/schema"
|
|
"github.com/rancher/steve/pkg/schema/converter"
|
|
apiextcontrollerv1 "github.com/rancher/wrangler/v3/pkg/generated/controllers/apiextensions.k8s.io/v1"
|
|
v1 "github.com/rancher/wrangler/v3/pkg/generated/controllers/apiregistration.k8s.io/v1"
|
|
"github.com/sirupsen/logrus"
|
|
"golang.org/x/sync/errgroup"
|
|
"golang.org/x/sync/semaphore"
|
|
authorizationv1 "k8s.io/api/authorization/v1"
|
|
apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
k8sapimachineryschema "k8s.io/apimachinery/pkg/runtime/schema"
|
|
"k8s.io/client-go/discovery"
|
|
authorizationv1client "k8s.io/client-go/kubernetes/typed/authorization/v1"
|
|
apiv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
|
)
|
|
|
|
var (
|
|
listPool = semaphore.NewWeighted(10)
|
|
typeNameChanges = map[string]string{
|
|
"extensions.v1beta1.ingress": "networking.k8s.io.v1beta1.ingress",
|
|
}
|
|
)
|
|
|
|
type SchemasHandlerFunc func(schemas *schema2.Collection, changedSchemas map[string]*types.APISchema, deletedSomething bool) error
|
|
|
|
func (s SchemasHandlerFunc) OnSchemas(schemas *schema2.Collection, changedSchemas map[string]*types.APISchema, forceChange bool) error {
|
|
return s(schemas, changedSchemas, forceChange)
|
|
}
|
|
|
|
type handler struct {
|
|
sync.Mutex
|
|
|
|
// refreshLock prevents refreshAll to be run in parallel
|
|
refreshLock sync.Mutex
|
|
|
|
ctx context.Context
|
|
toSync int32
|
|
schemas *schema2.Collection
|
|
client discovery.DiscoveryInterface
|
|
cols *common.DynamicColumns
|
|
crdClient apiextcontrollerv1.CustomResourceDefinitionClient
|
|
ssar authorizationv1client.SelfSubjectAccessReviewInterface
|
|
handler SchemasHandlerFunc
|
|
changedIDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
createdCRDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
deletedCRDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
apiServiceChanged bool
|
|
gvksFromKeys map[string][]k8sapimachineryschema.GroupVersionKind
|
|
}
|
|
|
|
func Register(ctx context.Context,
|
|
cols *common.DynamicColumns,
|
|
discovery discovery.DiscoveryInterface,
|
|
crd apiextcontrollerv1.CustomResourceDefinitionController,
|
|
apiService v1.APIServiceController,
|
|
ssar authorizationv1client.SelfSubjectAccessReviewInterface,
|
|
schemasHandler SchemasHandlerFunc,
|
|
schemas *schema2.Collection) {
|
|
|
|
h := &handler{
|
|
ctx: ctx,
|
|
cols: cols,
|
|
client: discovery,
|
|
schemas: schemas,
|
|
handler: schemasHandler,
|
|
crdClient: crd,
|
|
ssar: ssar,
|
|
changedIDs: make(map[k8sapimachineryschema.GroupVersionKind]bool),
|
|
gvksFromKeys: make(map[string][]k8sapimachineryschema.GroupVersionKind),
|
|
createdCRDs: make(map[k8sapimachineryschema.GroupVersionKind]bool),
|
|
deletedCRDs: make(map[k8sapimachineryschema.GroupVersionKind]bool),
|
|
}
|
|
|
|
apiService.OnChange(ctx, "schema", h.OnChangeAPIService)
|
|
crd.OnChange(ctx, "schema", h.OnChangeCRD)
|
|
}
|
|
|
|
func (h *handler) handleDeletedCRD(key string, crd *apiextv1.CustomResourceDefinition) {
|
|
h.Lock()
|
|
defer h.Unlock()
|
|
gvkList, ok := h.gvksFromKeys[key]
|
|
if !ok {
|
|
logrus.Infof("No associated GVK for CRD key %s", key)
|
|
return
|
|
}
|
|
for _, gvk := range gvkList {
|
|
h.deletedCRDs[gvk] = true
|
|
}
|
|
delete(h.gvksFromKeys, key) // Don't need this anymore
|
|
h.queueRefresh()
|
|
}
|
|
|
|
func (h *handler) OnChangeCRD(key string, crd *apiextv1.CustomResourceDefinition) (*apiextv1.CustomResourceDefinition, error) {
|
|
if crd == nil {
|
|
h.handleDeletedCRD(key, crd)
|
|
return crd, nil
|
|
}
|
|
spec := crd.Spec
|
|
group := spec.Group
|
|
kind := spec.Names.Kind
|
|
gvkList := make([]k8sapimachineryschema.GroupVersionKind, len(spec.Versions))
|
|
h.Lock()
|
|
defer h.Unlock()
|
|
for i, version := range spec.Versions {
|
|
gvk := k8sapimachineryschema.GroupVersionKind{Group: group, Version: version.Name, Kind: kind}
|
|
gvkList[i] = gvk
|
|
h.changedIDs[gvk] = true
|
|
_, ok := h.gvksFromKeys[key]
|
|
if !ok {
|
|
h.createdCRDs[gvk] = true
|
|
}
|
|
}
|
|
h.gvksFromKeys[key] = gvkList
|
|
h.queueRefresh()
|
|
return crd, nil
|
|
}
|
|
|
|
func (h *handler) OnChangeAPIService(key string, api *apiv1.APIService) (*apiv1.APIService, error) {
|
|
h.apiServiceChanged = true
|
|
h.queueRefresh()
|
|
return api, nil
|
|
}
|
|
|
|
func (h *handler) queueRefresh() {
|
|
atomic.StoreInt32(&h.toSync, 1)
|
|
|
|
go func() {
|
|
time.Sleep(500 * time.Millisecond)
|
|
var err error
|
|
var changedIDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
var deletedCRDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
var createdCRDs map[k8sapimachineryschema.GroupVersionKind]bool
|
|
var apiServiceChanged bool
|
|
h.Lock()
|
|
if len(h.createdCRDs) > 0 {
|
|
createdCRDs = h.createdCRDs
|
|
h.createdCRDs = make(map[k8sapimachineryschema.GroupVersionKind]bool)
|
|
}
|
|
if len(h.deletedCRDs) > 0 {
|
|
deletedCRDs = h.deletedCRDs
|
|
h.deletedCRDs = make(map[k8sapimachineryschema.GroupVersionKind]bool)
|
|
}
|
|
if len(h.changedIDs) > 0 {
|
|
changedIDs = h.changedIDs
|
|
h.changedIDs = make(map[k8sapimachineryschema.GroupVersionKind]bool)
|
|
}
|
|
if h.apiServiceChanged {
|
|
apiServiceChanged = true
|
|
h.apiServiceChanged = false
|
|
}
|
|
h.Unlock()
|
|
crdNumCountChanged := len(deletedCRDs) > 0 || len(createdCRDs) > 0
|
|
if len(changedIDs) > 0 || apiServiceChanged || crdNumCountChanged {
|
|
err = h.refreshAll(h.ctx, changedIDs, crdNumCountChanged)
|
|
}
|
|
if err != nil {
|
|
logrus.Errorf("failed to sync schemas: %v", err)
|
|
atomic.StoreInt32(&h.toSync, 1)
|
|
}
|
|
}()
|
|
}
|
|
|
|
func isListOrGetable(schema *types.APISchema) bool {
|
|
for _, verb := range attributes.Verbs(schema) {
|
|
switch verb {
|
|
case "list":
|
|
return true
|
|
case "get":
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func IsListWatchable(schema *types.APISchema) bool {
|
|
var (
|
|
canList bool
|
|
canWatch bool
|
|
)
|
|
|
|
for _, verb := range attributes.Verbs(schema) {
|
|
switch verb {
|
|
case "list":
|
|
canList = true
|
|
case "watch":
|
|
canWatch = true
|
|
}
|
|
}
|
|
|
|
return canList && canWatch
|
|
}
|
|
|
|
func (h *handler) getColumns(ctx context.Context, schemas map[string]*types.APISchema) error {
|
|
eg := errgroup.Group{}
|
|
|
|
for _, schema := range schemas {
|
|
if !isListOrGetable(schema) {
|
|
continue
|
|
}
|
|
|
|
if err := listPool.Acquire(ctx, 1); err != nil {
|
|
return err
|
|
}
|
|
|
|
s := schema
|
|
eg.Go(func() error {
|
|
defer listPool.Release(1)
|
|
return h.cols.SetColumns(ctx, s)
|
|
})
|
|
}
|
|
|
|
return eg.Wait()
|
|
}
|
|
|
|
func (h *handler) refreshAll(ctx context.Context, changedGVKs map[k8sapimachineryschema.GroupVersionKind]bool, forceChange bool) error {
|
|
h.refreshLock.Lock()
|
|
defer h.refreshLock.Unlock()
|
|
|
|
if !h.needToSync() {
|
|
return nil
|
|
}
|
|
|
|
schemas, err := converter.ToSchemas(h.crdClient, h.client)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
filteredSchemas := map[string]*types.APISchema{}
|
|
changedSchemasByID := map[string]*types.APISchema{}
|
|
for _, schema := range schemas {
|
|
if IsListWatchable(schema) {
|
|
if preferredTypeExists(schema, schemas) {
|
|
continue
|
|
}
|
|
if ok, err := h.allowed(ctx, schema); err != nil {
|
|
return err
|
|
} else if !ok {
|
|
continue
|
|
}
|
|
}
|
|
|
|
gvk := attributes.GVK(schema)
|
|
if gvk.Kind != "" {
|
|
gvr := attributes.GVR(schema)
|
|
schema.ID = converter.GVKToSchemaID(gvk)
|
|
schema.PluralName = converter.GVRToPluralName(gvr)
|
|
}
|
|
filteredSchemas[schema.ID] = schema
|
|
if changedGVKs[gvk] {
|
|
// nil[x] is always false if the first-time runner called this
|
|
changedSchemasByID[schema.ID] = schema
|
|
}
|
|
}
|
|
|
|
if err := h.getColumns(h.ctx, filteredSchemas); err != nil {
|
|
return err
|
|
}
|
|
|
|
h.schemas.Reset(filteredSchemas)
|
|
if h.handler != nil {
|
|
return h.handler.OnSchemas(h.schemas, changedSchemasByID, forceChange)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func preferredTypeExists(schema *types.APISchema, schemas map[string]*types.APISchema) bool {
|
|
if replacement, ok := typeNameChanges[schema.ID]; ok && schemas[replacement] != nil {
|
|
return true
|
|
}
|
|
pg := attributes.PreferredGroup(schema)
|
|
pv := attributes.PreferredVersion(schema)
|
|
if pg == "" && pv == "" {
|
|
return false
|
|
}
|
|
|
|
gvk := attributes.GVK(schema)
|
|
if pg != "" {
|
|
gvk.Group = pg
|
|
}
|
|
if pv != "" {
|
|
gvk.Version = pv
|
|
}
|
|
|
|
_, ok := schemas[converter.GVKToVersionedSchemaID(gvk)]
|
|
return ok
|
|
}
|
|
|
|
func (h *handler) allowed(ctx context.Context, schema *types.APISchema) (bool, error) {
|
|
gvr := attributes.GVR(schema)
|
|
ssar, err := h.ssar.Create(ctx, &authorizationv1.SelfSubjectAccessReview{
|
|
Spec: authorizationv1.SelfSubjectAccessReviewSpec{
|
|
ResourceAttributes: &authorizationv1.ResourceAttributes{
|
|
Verb: "list",
|
|
Group: gvr.Group,
|
|
Version: gvr.Version,
|
|
Resource: gvr.Resource,
|
|
},
|
|
},
|
|
}, metav1.CreateOptions{})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return ssar.Status.Allowed && !ssar.Status.Denied, nil
|
|
}
|
|
|
|
func (h *handler) needToSync() bool {
|
|
old := atomic.SwapInt32(&h.toSync, 0)
|
|
return old == 1
|
|
}
|