Skip to content

Commit

Permalink
patch: update the primarykinds map to be a slice of GroupVersionKinds
Browse files Browse the repository at this point in the history 
The main issue with the current code is, it uses a GroupVersionKinds
cache and that a simple `map[string]GroupVersionKind` and it does not
allow multiple versions of the same Kind.

This patch changes this logic the way, the map stores a list of
GroupVersionKinds. The affects 2 endpoints: GetObject and ListObjects.

With this patch:
1. GetObject will iterate over all known Kind versions and return with
   the first found object and skips the rest.
2. ListObjects iterates over all known Kind versions and aggregates the
   results list into a single list and returns with that list.

Signed-off-by: Balazs Nadasdi <[email protected]>
  • Loading branch information
@yitsushi
yitsushi committed Jul 3, 2023
1 parent 87da4e3 commit c5583d7
Show file tree
Hide file tree
Showing 3 changed files with 134 additions and 115 deletions.
6 changes: 3 additions & 3 deletions core/server/events.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,13 +40,13 @@ func (cs *coreServer) ListEvents(ctx context.Context, msg *pb.ListEventsRequest)

kind := msg.InvolvedObject.Kind

gvk, err := cs.primaryKinds.Lookup(kind)
if err != nil {
gvks, err := cs.primaryKinds.Lookup(kind)
if err != nil || len(gvks) == 0 {
return nil, status.Errorf(codes.InvalidArgument, "bad request: not a recognized object kind")
}

fields := client.MatchingFields{
"involvedObject.kind": gvk.Kind,
"involvedObject.kind": gvks[0].Kind,
"involvedObject.name": msg.InvolvedObject.Name,
"involvedObject.namespace": msg.InvolvedObject.Namespace,
}
Expand Down
207 changes: 105 additions & 102 deletions core/server/objects.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
v1 "k8s.io/api/apps/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"sigs.k8s.io/controller-runtime/pkg/client"
)

Expand All @@ -41,14 +42,11 @@ func getUnstructuredHelmReleaseInventory(ctx context.Context, obj unstructured.U
}

func (cs *coreServer) ListObjects(ctx context.Context, msg *pb.ListObjectsRequest) (*pb.ListObjectsResponse, error) {
respErrors := []*pb.ListError{}

gvk, err := cs.primaryKinds.Lookup(msg.Kind)
if err != nil {
return nil, err
}

var err error
var clustersClient clustersmngr.Client
var results []*pb.Object

respErrors := []*pb.ListError{}

if msg.ClusterName != "" {
clustersClient, err = cs.clustersManager.GetImpersonatedClientForCluster(ctx, auth.Principal(ctx), msg.ClusterName)
Expand All @@ -66,92 +64,93 @@ func (cs *coreServer) ListObjects(ctx context.Context, msg *pb.ListObjectsReques
}
}

clist := clustersmngr.NewClusteredList(func() client.ObjectList {
list := unstructured.UnstructuredList{}
list.SetGroupVersionKind(*gvk)
return &list
})

listOptions := []client.ListOption{
client.InNamespace(msg.Namespace),
}
if len(msg.Labels) > 0 {
listOptions = append(listOptions, client.MatchingLabels(msg.Labels))
}
err = cs.primaryKinds.RunWithKindVersions(msg.Kind, func(gvk schema.GroupVersionKind) error {
clist := clustersmngr.NewClusteredList(func() client.ObjectList {
list := unstructured.UnstructuredList{}
list.SetGroupVersionKind(gvk)
return &list
})

if err := clustersClient.ClusteredList(ctx, clist, true, listOptions...); err != nil {
var errs clustersmngr.ClusteredListError
if !errors.As(err, &errs) {
return nil, err
listOptions := []client.ListOption{
client.InNamespace(msg.Namespace),
}

for _, e := range errs.Errors {
respErrors = append(respErrors, &pb.ListError{ClusterName: e.Cluster, Namespace: e.Namespace, Message: e.Err.Error()})
if len(msg.Labels) > 0 {
listOptions = append(listOptions, client.MatchingLabels(msg.Labels))
}
}

var results []*pb.Object

clusterUserNamespaces := cs.clustersManager.GetUserNamespaces(auth.Principal(ctx))

for n, lists := range clist.Lists() {
for _, l := range lists {
list, ok := l.(*unstructured.UnstructuredList)
if !ok {
continue
if err := clustersClient.ClusteredList(ctx, clist, true, listOptions...); err != nil {
var errs clustersmngr.ClusteredListError
if !errors.As(err, &errs) {
return err
}

for _, unstructuredObj := range list.Items {
tenant := GetTenant(unstructuredObj.GetNamespace(), n, clusterUserNamespaces)
for _, e := range errs.Errors {
respErrors = append(respErrors, &pb.ListError{ClusterName: e.Cluster, Namespace: e.Namespace, Message: e.Err.Error()})
}
}

var obj client.Object = &unstructuredObj
clusterUserNamespaces := cs.clustersManager.GetUserNamespaces(auth.Principal(ctx))

var inventory []*pb.GroupVersionKind = nil
var info string
for n, lists := range clist.Lists() {
for _, l := range lists {
list, ok := l.(*unstructured.UnstructuredList)
if !ok {
continue
}

switch gvk.Kind {
case "Secret":
obj, err = sanitizeSecret(&unstructuredObj)
if err != nil {
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: fmt.Sprintf("error sanitizing secrets: %v", err)})
continue
for _, unstructuredObj := range list.Items {
tenant := GetTenant(unstructuredObj.GetNamespace(), n, clusterUserNamespaces)

var obj client.Object = &unstructuredObj

var inventory []*pb.GroupVersionKind = nil
var info string

switch gvk.Kind {
case "Secret":
obj, err = sanitizeSecret(&unstructuredObj)
if err != nil {
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: fmt.Sprintf("error sanitizing secrets: %v", err)})
continue
}
case v2beta1.HelmReleaseKind:
inventory, err = getUnstructuredHelmReleaseInventory(ctx, unstructuredObj, clustersClient, n)
if err != nil {
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: err.Error()})
inventory = nil // We can still display most things without inventory

cs.logger.V(logger.LogLevelDebug).Info("Couldn't grab inventory for helm release", "error", err)
}
case "StatefulSet":
clusterName, kind, err := parseSessionInfo(unstructuredObj)
if err != nil {
break
}

created, _ := cs.sessionObjectsCreated(ctx, clusterName, "flux-system", kind)

if created {
info = sessionObjectsInfo
}
}
case v2beta1.HelmReleaseKind:
inventory, err = getUnstructuredHelmReleaseInventory(ctx, unstructuredObj, clustersClient, n)
if err != nil {
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: err.Error()})
inventory = nil // We can still display most things without inventory

cs.logger.V(logger.LogLevelDebug).Info("Couldn't grab inventory for helm release", "error", err)
}
case "StatefulSet":
clusterName, kind, err := parseSessionInfo(unstructuredObj)
o, err := types.K8sObjectToProto(obj, n, tenant, inventory, info)
if err != nil {
break
}

created, _ := cs.sessionObjectsCreated(ctx, clusterName, "flux-system", kind)

if created {
info = sessionObjectsInfo
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: "converting items: " + err.Error()})
continue
}
}

o, err := types.K8sObjectToProto(obj, n, tenant, inventory, info)
if err != nil {
respErrors = append(respErrors, &pb.ListError{ClusterName: n, Message: "converting items: " + err.Error()})
continue
results = append(results, o)
}

results = append(results, o)
}
}
}
return nil
})

return &pb.ListObjectsResponse{
Objects: results,
Errors: respErrors,
}, nil
}, err
}

func parseSessionInfo(unstructuredObj unstructured.Unstructured) (string, string, error) {
Expand Down Expand Up @@ -225,51 +224,55 @@ func (cs *coreServer) GetObject(ctx context.Context, msg *pb.GetObjectRequest) (
return nil, fmt.Errorf("error getting impersonating client: %w", err)
}

gvk, err := cs.primaryKinds.Lookup(msg.Kind)
gvks, err := cs.primaryKinds.Lookup(msg.Kind)
if err != nil {
return nil, err
}

unstructuredObj := unstructured.Unstructured{}
unstructuredObj.SetGroupVersionKind(*gvk)
for _, gvk := range gvks {
unstructuredObj := unstructured.Unstructured{}
unstructuredObj.SetGroupVersionKind(gvk)

key := client.ObjectKey{
Name: msg.Name,
Namespace: msg.Namespace,
}
key := client.ObjectKey{
Name: msg.Name,
Namespace: msg.Namespace,
}

if err := clustersClient.Get(ctx, msg.ClusterName, key, &unstructuredObj); err != nil {
return nil, err
}
if err := clustersClient.Get(ctx, msg.ClusterName, key, &unstructuredObj); err != nil {
continue
}

var inventory []*pb.GroupVersionKind = nil
var inventory []*pb.GroupVersionKind = nil

var obj client.Object = &unstructuredObj
var obj client.Object = &unstructuredObj

switch gvk.Kind {
case "Secret":
obj, err = sanitizeSecret(&unstructuredObj)
if err != nil {
return nil, fmt.Errorf("error sanitizing secrets: %w", err)
}
case v2beta1.HelmReleaseKind:
inventory, err = getUnstructuredHelmReleaseInventory(ctx, unstructuredObj, clustersClient, msg.ClusterName)
if err != nil {
inventory = nil // We can still display most things without inventory
switch gvk.Kind {
case "Secret":
obj, err = sanitizeSecret(&unstructuredObj)
if err != nil {
return nil, fmt.Errorf("error sanitizing secrets: %w", err)
}
case v2beta1.HelmReleaseKind:
inventory, err = getUnstructuredHelmReleaseInventory(ctx, unstructuredObj, clustersClient, msg.ClusterName)
if err != nil {
inventory = nil // We can still display most things without inventory

cs.logger.V(logger.LogLevelDebug).Info("Couldn't grab inventory for helm release", "error", err)
cs.logger.V(logger.LogLevelDebug).Info("Couldn't grab inventory for helm release", "error", err)
}
}
}

clusterUserNamespaces := cs.clustersManager.GetUserNamespaces(auth.Principal(ctx))
clusterUserNamespaces := cs.clustersManager.GetUserNamespaces(auth.Principal(ctx))

tenant := GetTenant(obj.GetNamespace(), msg.ClusterName, clusterUserNamespaces)
tenant := GetTenant(obj.GetNamespace(), msg.ClusterName, clusterUserNamespaces)

res, err := types.K8sObjectToProto(obj, msg.ClusterName, tenant, inventory, "")
res, err := types.K8sObjectToProto(obj, msg.ClusterName, tenant, inventory, "")

if err != nil {
return nil, fmt.Errorf("converting object to proto: %w", err)
if err != nil {
return nil, fmt.Errorf("converting object to proto: %w", err)
}

return &pb.GetObjectResponse{Object: res}, nil
}

return &pb.GetObjectResponse{Object: res}, nil
return nil, fmt.Errorf("%s/%s not found with any registered GroupVersions", msg.Kind, msg.Name)
}
36 changes: 26 additions & 10 deletions core/server/primarykinds.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,12 @@ import (
)

type PrimaryKinds struct {
kinds map[string]schema.GroupVersionKind
kinds map[string][]schema.GroupVersionKind
}

func New() *PrimaryKinds {
kinds := PrimaryKinds{}
kinds.kinds = make(map[string]schema.GroupVersionKind)
kinds.kinds = make(map[string][]schema.GroupVersionKind)

return &kinds
}
Expand All @@ -27,7 +27,7 @@ func DefaultPrimaryKinds() (*PrimaryKinds, error) {
}

for gvk := range scheme.AllKnownTypes() {
kinds.kinds[gvk.Kind] = gvk
kinds.kinds[gvk.Kind] = append(kinds.kinds[gvk.Kind], gvk)
}

return kinds, nil
Expand All @@ -37,23 +37,39 @@ func DefaultPrimaryKinds() (*PrimaryKinds, error) {
// This errors if the kind is already set, as this likely indicates 2
// different uses for the same kind string.
func (pk *PrimaryKinds) Add(kind string, gvk schema.GroupVersionKind) error {
_, ok := pk.kinds[kind]
if ok {
return fmt.Errorf("couldn't add kind %v - already added", kind)
for _, version := range pk.kinds[kind] {
if version.Version == gvk.Version {
return fmt.Errorf("couldn't add kind %v with version %s - already added", kind, gvk.Version)
}
}

pk.kinds[kind] = gvk
pk.kinds[kind] = append(pk.kinds[kind], gvk)

return nil
}

// Lookup ensures that a kind name is known, white-listed, and returns
// the full GVK for that kind
func (pk *PrimaryKinds) Lookup(kind string) (*schema.GroupVersionKind, error) {
gvk, ok := pk.kinds[kind]
func (pk *PrimaryKinds) Lookup(kind string) ([]schema.GroupVersionKind, error) {
gvks, ok := pk.kinds[kind]
if !ok {
return nil, fmt.Errorf("looking up objects of kind %v not supported", kind)
}

return &gvk, nil
return gvks, nil
}

func (pk *PrimaryKinds) RunWithKindVersions(kind string, fn func(gvk schema.GroupVersionKind) error) error {
kindVersions, err := pk.Lookup(kind)
if err != nil {
return err
}

for _, gvk := range kindVersions {
if err := fn(gvk); err != nil {
return err
}
}

return nil
}

0 comments on commit c5583d7

Please sign in to comment.