Merge pull request #308 from jetstack/refactoring-dg-IV

Refactor k8d/dynamic datagatherer

The k8s/dynamic is now going to use a sharedIndexinformer for k8s native
v1 resources, instead of using a dynamicinformer. This reduces the
current memory usage by about 25MB.

Author: Oluwole Fadeyi

pkg/datagatherer/k8s/cache.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/jetstack/preflight/api"
 	"github.com/pmylund/go-cache"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 // time interface, this is used to fetch the current time
@@ -24,60 +24,52 @@ func (*realTime) now() time.Time {
 	return time.Now()
 }
 
+type cacheResource interface {
+	GetUID() types.UID
+	GetNamespace() string
+}
+
 // onAdd handles the informer creation events, adding the created runtime.Object
 // to the data gatherer's cache. The cache key is the uid of the object
 func onAdd(obj interface{}, dgCache *cache.Cache) {
-	item := obj.(*unstructured.Unstructured)
-	if metadata, ok := item.Object["metadata"]; ok {
-		data := metadata.(map[string]interface{})
-		if uid, ok := data["uid"]; ok {
-			cacheObject := &api.GatheredResource{
-				Resource: obj,
-			}
-			dgCache.Set(uid.(string), cacheObject, cache.DefaultExpiration)
-		} else {
-			log.Printf("could not %q resource %q to the cache, missing uid field", "add", data["name"].(string))
+	item, ok := obj.(cacheResource)
+	if ok {
+		cacheObject := &api.GatheredResource{
+			Resource: obj,
 		}
-	} else {
-		log.Printf("could not %q resource to the cache, missing metadata", "add")
+		dgCache.Set(string(item.GetUID()), cacheObject, cache.DefaultExpiration)
+		return
 	}
+	log.Printf("could not %q resource to the cache, missing metadata/uid field", "add")
+
 }
 
 // onUpdate handles the informer update events, replacing the old object with the new one
 // if it's present in the data gatherer's cache, (if the object isn't present, it gets added).
 // The cache key is the uid of the object
 func onUpdate(old, new interface{}, dgCache *cache.Cache) {
-	item := old.(*unstructured.Unstructured)
-	if metadata, ok := item.Object["metadata"]; ok {
-		data := metadata.(map[string]interface{})
-		if uid, ok := data["uid"]; ok {
-			cacheObject := updateCacheGatheredResource(uid.(string), new, dgCache)
-			dgCache.Set(uid.(string), cacheObject, cache.DefaultExpiration)
-		} else {
-			log.Printf("could not %q resource %q to the cache, missing uid field", "update", data["name"].(string))
-		}
-	} else {
-		log.Printf("could not %q resource to the cache, missing metadata", "update")
+	item, ok := old.(cacheResource)
+	if ok {
+		cacheObject := updateCacheGatheredResource(string(item.GetUID()), new, dgCache)
+		dgCache.Set(string(item.GetUID()), cacheObject, cache.DefaultExpiration)
+		return
 	}
+
+	log.Printf("could not %q resource to the cache, missing metadata/uid field", "update")
 }
 
 // onDelete handles the informer deletion events, updating the object's properties with the deletion
 // time of the object (but not removing the object from the cache).
 // The cache key is the uid of the object
 func onDelete(obj interface{}, dgCache *cache.Cache) {
-	item := obj.(*unstructured.Unstructured)
-	if metadata, ok := item.Object["metadata"]; ok {
-		data := metadata.(map[string]interface{})
-		if uid, ok := data["uid"]; ok {
-			cacheObject := updateCacheGatheredResource(uid.(string), obj, dgCache)
-			cacheObject.DeletedAt = api.Time{Time: clock.now()}
-			dgCache.Set(uid.(string), cacheObject, cache.DefaultExpiration)
-		} else {
-			log.Printf("could not %q resource %q to the cache, missing uid field", "delete", data["name"].(string))
-		}
-	} else {
-		log.Printf("could not %q resource to the cache, missing metadata", "delete")
+	item, ok := obj.(cacheResource)
+	if ok {
+		cacheObject := updateCacheGatheredResource(string(item.GetUID()), obj, dgCache)
+		cacheObject.DeletedAt = api.Time{Time: clock.now()}
+		dgCache.Set(string(item.GetUID()), cacheObject, cache.DefaultExpiration)
+		return
 	}
+	log.Printf("could not %q resource to the cache, missing metadata/uid field", "delete")
 }
 
 // creates a new updated instance of a cache object, with the resource

pkg/datagatherer/k8s/client.go

@@ -5,6 +5,7 @@ import (
 	"github.com/pkg/errors"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 )
@@ -43,6 +44,22 @@ func NewDiscoveryClient(kubeconfigPath string) (discovery.DiscoveryClient, error
 	return *discoveryClient, nil
 }
 
+// NewClientSet creates a new kubernetes clientset using the provided kubeconfig.
+// If kubeconfigPath is not set/empty, it will attempt to load configuration using
+// the default loading rules.
+func NewClientSet(kubeconfigPath string) (kubernetes.Interface, error) {
+	var clientset *kubernetes.Clientset
+	cfg, err := loadRESTConfig(kubeconfigPath)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	clientset, err = kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return clientset, nil
+}
+
 func loadRESTConfig(path string) (*rest.Config, error) {
 	switch path {
 	// If the kubeconfig path is not provided, use the default loading rules