replace kubeclient & add dynamic client

client/certificates.go

@@ -0,0 +1,205 @@
+package client
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nirmata/kube-policy/utils"
+
+	certificates "k8s.io/api/certificates/v1beta1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// Issues TLS certificate for webhook server using given PEM private key
+// Returns signed and approved TLS certificate in PEM format
+func (c *Client) GenerateTlsPemPair(props utils.TlsCertificateProps) (*utils.TlsPemPair, error) {
+	privateKey, err := utils.TlsGeneratePrivateKey()
+	if err != nil {
+		return nil, err
+	}
+
+	certRequest, err := utils.TlsCertificateGenerateRequest(privateKey, props)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Unable to create certificate request: %v", err))
+	}
+
+	certRequest, err = c.submitAndApproveCertificateRequest(certRequest)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Unable to submit and approve certificate request: %v", err))
+	}
+
+	tlsCert, err := c.fetchCertificateFromRequest(certRequest, 10)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Unable to fetch certificate from request: %v", err))
+	}
+
+	return &utils.TlsPemPair{
+		Certificate: tlsCert,
+		PrivateKey:  utils.TlsPrivateKeyToPem(privateKey),
+	}, nil
+}
+
+// Submits and approves certificate request, returns request which need to be fetched
+func (c *Client) submitAndApproveCertificateRequest(req *certificates.CertificateSigningRequest) (*certificates.CertificateSigningRequest, error) {
+	//TODO: using the CSR interface from the kubeclient
+	certClient, err := c.GetCSRInterface()
+	if err != nil {
+		return nil, err
+	}
+	//	certClient := kc.client.CertificatesV1beta1().CertificateSigningRequests()
+	csrList, err := c.ListResource("certificatesigningrequest", "")
+	//	csrList, err := certClient.List(metav1.ListOptions{})
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Unable to list existing certificate requests: %v", err))
+	}
+
+	for _, csr := range csrList.Items {
+		csr.GetName()
+		if csr.GetName() == req.ObjectMeta.Name {
+			// Delete
+			err := c.DeleteResouce("certificatesigningrequest", "", csr.GetName())
+			if err != nil {
+				return nil, errors.New(fmt.Sprintf("Unable to delete existing certificate request: %v", err))
+			}
+			c.logger.Printf("Old certificate request is deleted")
+			break
+		}
+	}
+
+	// Create
+	unstrRes, err := c.CreateResource("certificatesigningrequest", "", req)
+	//	res, err := certClient.Create(req)
+	if err != nil {
+		return nil, err
+	}
+	c.logger.Printf("Certificate request %s is created", unstrRes.GetName())
+
+	res, err := convertToCSR(unstrRes)
+	if err != nil {
+		return nil, err
+	}
+	res.Status.Conditions = append(res.Status.Conditions, certificates.CertificateSigningRequestCondition{
+		Type:    certificates.CertificateApproved,
+		Reason:  "NKP-Approve",
+		Message: "This CSR was approved by Nirmata kube-policy controller",
+	})
+	res, err = certClient.UpdateApproval(res)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("Unable to approve certificate request: %v", err))
+	}
+	c.logger.Printf("Certificate request %s is approved", res.ObjectMeta.Name)
+
+	return res, nil
+}
+
+const certificateFetchWaitInterval time.Duration = 200 * time.Millisecond
+
+// Fetches certificate from given request. Tries to obtain certificate for maxWaitSeconds
+func (c *Client) fetchCertificateFromRequest(req *certificates.CertificateSigningRequest, maxWaitSeconds uint8) ([]byte, error) {
+	// TODO: react of SIGINT and SIGTERM
+	timeStart := time.Now()
+	c.GetResource("certificatesigningrequest", "", req.ObjectMeta.Name)
+	for time.Now().Sub(timeStart) < time.Duration(maxWaitSeconds)*time.Second {
+		unstrR, err := c.GetResource("certificatesigningrequest", "", req.ObjectMeta.Name)
+		if err != nil {
+			return nil, err
+		}
+		r, err := convertToCSR(unstrR)
+		if err != nil {
+			return nil, err
+		}
+
+		if r.Status.Certificate != nil {
+			return r.Status.Certificate, nil
+		}
+
+		for _, condition := range r.Status.Conditions {
+			if condition.Type == certificates.CertificateDenied {
+				return nil, errors.New(condition.String())
+			}
+		}
+	}
+	return nil, errors.New(fmt.Sprintf("Cerificate fetch timeout is reached: %d seconds", maxWaitSeconds))
+}
+
+const privateKeyField string = "privateKey"
+const certificateField string = "certificate"
+
+// Reads the pair of TLS certificate and key from the specified secret.
+func (c *Client) ReadTlsPair(props utils.TlsCertificateProps) *utils.TlsPemPair {
+	name := generateSecretName(props)
+	unstrSecret, err := c.GetResource("secrets", props.Namespace, name)
+	if err != nil {
+		c.logger.Printf("Unable to get secret %s/%s: %s", props.Namespace, name, err)
+		return nil
+	}
+	secret, err := convertToSecret(unstrSecret)
+	if err != nil {
+		return nil
+	}
+	pemPair := utils.TlsPemPair{
+		Certificate: secret.Data[certificateField],
+		PrivateKey:  secret.Data[privateKeyField],
+	}
+	if len(pemPair.Certificate) == 0 {
+		c.logger.Printf("TLS Certificate not found in secret %s/%s", props.Namespace, name)
+		return nil
+	}
+	if len(pemPair.PrivateKey) == 0 {
+		c.logger.Printf("TLS PrivateKey not found in secret %s/%s", props.Namespace, name)
+		return nil
+	}
+	return &pemPair
+}
+
+// Writes the pair of TLS certificate and key to the specified secret.
+// Updates existing secret or creates new one.
+func (c *Client) WriteTlsPair(props utils.TlsCertificateProps, pemPair *utils.TlsPemPair) error {
+	name := generateSecretName(props)
+	unstrSecret, err := c.GetResource("secrets", props.Namespace, name)
+	if err == nil { // Update existing secret
+		secret, err := convertToSecret(unstrSecret)
+		if err != nil {
+			return nil
+		}
+
+		if secret.Data == nil {
+			secret.Data = make(map[string][]byte)
+		}
+		secret.Data[certificateField] = pemPair.Certificate
+		secret.Data[privateKeyField] = pemPair.PrivateKey
+		c.UpdateResource("secrets", props.Namespace, secret)
+		if err == nil {
+			c.logger.Printf("Secret %s is updated", name)
+		}
+
+	} else { // Create new secret
+
+		secret := &v1.Secret{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "Secret",
+				APIVersion: "v1",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      name,
+				Namespace: props.Namespace,
+			},
+			Data: map[string][]byte{
+				certificateField: pemPair.Certificate,
+				privateKeyField:  pemPair.PrivateKey,
+			},
+		}
+
+		_, err := c.CreateResource("secrets", props.Namespace, secret)
+		if err == nil {
+			c.logger.Printf("Secret %s is created", name)
+		}
+	}
+	return err
+}
+
+func generateSecretName(props utils.TlsCertificateProps) string {
+	return utils.GenerateInClusterServiceName(props) + ".kube-policy-tls-pair"
+}

client/client.go

@@ -0,0 +1,292 @@
+package client
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/nirmata/kube-policy/config"
+	types "github.com/nirmata/kube-policy/pkg/apis/policy/v1alpha1"
+	apps "k8s.io/api/apps/v1"
+	certificates "k8s.io/api/certificates/v1beta1"
+	v1 "k8s.io/api/core/v1"
+	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/dynamic"
+	csrtype "k8s.io/client-go/kubernetes/typed/certificates/v1beta1"
+	event "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/rest"
+)
+
+type Client struct {
+	logger       *log.Logger
+	client       dynamic.Interface
+	clientConfig *rest.Config
+}
+
+func NewDynamicClient(config *rest.Config, logger *log.Logger) (*Client, error) {
+	client, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return &Client{
+		logger:       logger,
+		client:       client,
+		clientConfig: config,
+	}, nil
+}
+
+func (c *Client) GetKubePolicyDeployment() (*apps.Deployment, error) {
+	kubePolicyDeployment, err := c.GetResource("deployments", config.KubePolicyNamespace, config.KubePolicyDeploymentName)
+	if err != nil {
+		return nil, err
+	}
+	deploy := apps.Deployment{}
+	if err = runtime.DefaultUnstructuredConverter.FromUnstructured(kubePolicyDeployment.UnstructuredContent(), &deploy); err != nil {
+		return nil, err
+	}
+	return &deploy, nil
+}
+
+//TODO: can we use dynamic client to fetch the typed interface
+// or generate a kube client value to access the interface
+//GetEventsInterface provides typed interface for events
+func (c *Client) GetEventsInterface() (event.EventInterface, error) {
+	kubeClient, err := newKubeClient(c.clientConfig)
+	if err != nil {
+		return nil, err
+	}
+	return kubeClient.CoreV1().Events(""), nil
+}
+
+func (c *Client) GetCSRInterface() (csrtype.CertificateSigningRequestInterface, error) {
+	kubeClient, err := newKubeClient(c.clientConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return kubeClient.CertificatesV1beta1().CertificateSigningRequests(), nil
+}
+
+func (c *Client) getInterface(kind string) dynamic.NamespaceableResourceInterface {
+	return c.client.Resource(c.getGroupVersionMapper(kind))
+}
+
+func (c *Client) getResourceInterface(kind string, namespace string) dynamic.ResourceInterface {
+	// Get the resource interface
+	namespaceableInterface := c.getInterface(kind)
+	// Get the namespacable interface
+	var resourceInteface dynamic.ResourceInterface
+	if namespace != "" {
+		resourceInteface = namespaceableInterface.Namespace(namespace)
+	} else {
+		resourceInteface = namespaceableInterface
+	}
+	return resourceInteface
+}
+
+// Keep this a stateful as the resource list will be based on the kubernetes version we connect to
+func (c *Client) getGroupVersionMapper(kind string) schema.GroupVersionResource {
+	//TODO: add checks to see if the kind is supported
+	//TODO: build the resource list dynamically( by querying the registered resource kinds)
+	//TODO: the error scenarios
+	return getGrpVersionMapper(kind, c.clientConfig, false)
+}
+
+// GetResource returns the resource in unstructured/json format
+func (c *Client) GetResource(kind string, namespace string, name string) (*unstructured.Unstructured, error) {
+	return c.getResourceInterface(kind, namespace).Get(name, meta.GetOptions{})
+}
+
+// ListResource returns the list of resources in unstructured/json format
+// Access items using []Items
+func (c *Client) ListResource(kind string, namespace string) (*unstructured.UnstructuredList, error) {
+	return c.getResourceInterface(kind, namespace).List(meta.ListOptions{})
+}
+
+func (c *Client) DeleteResouce(kind string, namespace string, name string) error {
+	return c.getResourceInterface(kind, namespace).Delete(name, &meta.DeleteOptions{})
+
+}
+
+// CreateResource creates object for the specified kind/namespace
+func (c *Client) CreateResource(kind string, namespace string, obj interface{}) (*unstructured.Unstructured, error) {
+	// convert typed to unstructured obj
+	if unstructuredObj := convertToUnstructured(obj); unstructuredObj != nil {
+		return c.getResourceInterface(kind, namespace).Create(unstructuredObj, meta.CreateOptions{})
+	}
+	return nil, fmt.Errorf("Unable to create resource ")
+}
+
+// UpdateResource updates object for the specified kind/namespace
+func (c *Client) UpdateResource(kind string, namespace string, obj interface{}) (*unstructured.Unstructured, error) {
+	// convert typed to unstructured obj
+	if unstructuredObj := convertToUnstructured(obj); unstructuredObj != nil {
+		return c.getResourceInterface(kind, namespace).Update(unstructuredObj, meta.UpdateOptions{})
+	}
+	return nil, fmt.Errorf("Unable to update resource ")
+}
+
+func convertToUnstructured(obj interface{}) *unstructured.Unstructured {
+	unstructuredObj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("Unable to convert : %v", err))
+		return nil
+	}
+	return &unstructured.Unstructured{Object: unstructuredObj}
+}
+
+//ConvertToRuntimeObject converts unstructed to runtime.Object runtime instance
+func ConvertToRuntimeObject(obj *unstructured.Unstructured) (*runtime.Object, error) {
+	scheme := runtime.NewScheme()
+	gvk := obj.GroupVersionKind()
+	runtimeObj, err := scheme.New(gvk)
+	if err != nil {
+		return nil, err
+	}
+	if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &runtimeObj); err != nil {
+		return nil, err
+	}
+	return &runtimeObj, nil
+}
+
+//TODO: make this generic for all resource type
+//GenerateSecret to generate secrets
+func (c *Client) GenerateSecret(generator types.PolicyConfigGenerator, namespace string) error {
+	c.logger.Printf("Preparing to create secret %s/%s", namespace, generator.Name)
+	secret := &v1.Secret{}
+
+	if generator.CopyFrom != nil {
+		c.logger.Printf("Copying data from secret %s/%s", generator.CopyFrom.Namespace, generator.CopyFrom.Name)
+		// Get configMap resource
+		unstrSecret, err := c.GetResource("secret", generator.CopyFrom.Namespace, generator.CopyFrom.Name)
+		if err != nil {
+			return err
+		}
+		// typed object
+		secret, err = convertToSecret(unstrSecret)
+		if err != nil {
+			return err
+		}
+	}
+
+	secret.ObjectMeta = meta.ObjectMeta{
+		Name:      generator.Name,
+		Namespace: namespace,
+	}
+
+	// Copy data from generator to the new secret
+	if generator.Data != nil {
+		if secret.Data == nil {
+			secret.Data = make(map[string][]byte)
+		}
+
+		for k, v := range generator.Data {
+			secret.Data[k] = []byte(v)
+		}
+	}
+
+	go c.createSecretAfterNamespaceIsCreated(*secret, namespace)
+	return nil
+}
+
+//TODO: make this generic for all resource type
+//GenerateConfigMap to generate configMap
+func (c *Client) GenerateConfigMap(generator types.PolicyConfigGenerator, namespace string) error {
+	c.logger.Printf("Preparing to create configmap %s/%s", namespace, generator.Name)
+	configMap := &v1.ConfigMap{}
+
+	if generator.CopyFrom != nil {
+		c.logger.Printf("Copying data from configmap %s/%s", generator.CopyFrom.Namespace, generator.CopyFrom.Name)
+		// Get configMap resource
+		unstrConfigMap, err := c.GetResource("configmap", generator.CopyFrom.Namespace, generator.CopyFrom.Name)
+		if err != nil {
+			return err
+		}
+		// typed object
+		configMap, err = convertToConfigMap(unstrConfigMap)
+		if err != nil {
+			return err
+		}
+
+	}
+	configMap.ObjectMeta = meta.ObjectMeta{
+		Name:      generator.Name,
+		Namespace: namespace,
+	}
+
+	// Copy data from generator to the new configmap
+	if generator.Data != nil {
+		if configMap.Data == nil {
+			configMap.Data = make(map[string]string)
+		}
+
+		for k, v := range generator.Data {
+			configMap.Data[k] = v
+		}
+	}
+	go c.createConfigMapAfterNamespaceIsCreated(*configMap, namespace)
+	return nil
+}
+
+func convertToConfigMap(obj *unstructured.Unstructured) (*v1.ConfigMap, error) {
+	configMap := v1.ConfigMap{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &configMap); err != nil {
+		return nil, err
+	}
+	return &configMap, nil
+}
+
+func convertToSecret(obj *unstructured.Unstructured) (*v1.Secret, error) {
+	secret := v1.Secret{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &secret); err != nil {
+		return nil, err
+	}
+	return &secret, nil
+}
+
+func convertToCSR(obj *unstructured.Unstructured) (*certificates.CertificateSigningRequest, error) {
+	csr := certificates.CertificateSigningRequest{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &csr); err != nil {
+		return nil, err
+	}
+	return &csr, nil
+}
+
+func (c *Client) createConfigMapAfterNamespaceIsCreated(configMap v1.ConfigMap, namespace string) {
+	err := c.waitUntilNamespaceIsCreated(namespace)
+	if err == nil {
+		_, err = c.CreateResource("configmap", namespace, configMap)
+	}
+	if err != nil {
+		c.logger.Printf("Can't create a configmap: %s", err)
+	}
+}
+
+func (c *Client) createSecretAfterNamespaceIsCreated(secret v1.Secret, namespace string) {
+	err := c.waitUntilNamespaceIsCreated(namespace)
+	if err == nil {
+		_, err = c.CreateResource("secrets", namespace, secret)
+	}
+	if err != nil {
+		c.logger.Printf("Can't create a secret: %s", err)
+	}
+}
+
+// Waits until namespace is created with maximum duration maxWaitTimeForNamespaceCreation
+func (c *Client) waitUntilNamespaceIsCreated(name string) error {
+	timeStart := time.Now()
+
+	var lastError error = nil
+	for time.Now().Sub(timeStart) < namespaceCreationMaxWaitTime {
+		_, lastError = c.GetResource("namespace", "", name)
+		if lastError == nil {
+			break
+		}
+		time.Sleep(namespaceCreationWaitInterval)
+	}
+	return lastError
+}

client/client_test.go

@@ -0,0 +1,6 @@
+package client
+
+// GetResource
+// ListResource
+// CreateResource
+// getGroupVersionMapper (valid and invalid resources)

client/utils.go

@@ -0,0 +1,79 @@
+package client
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+const namespaceCreationMaxWaitTime time.Duration = 30 * time.Second
+const namespaceCreationWaitInterval time.Duration = 100 * time.Millisecond
+
+var groupVersionMapper map[string]schema.GroupVersionResource
+
+func getGrpVersionMapper(kind string, clientConfig *rest.Config, refresh bool) schema.GroupVersionResource {
+	grpVersionSchema := schema.GroupVersionResource{}
+
+	if groupVersionMapper == nil || refresh {
+		groupVersionMapper = make(map[string]schema.GroupVersionResource)
+		// refesh the mapper
+		if err := refreshRegisteredResources(groupVersionMapper, clientConfig); err != nil {
+			utilruntime.HandleError(err)
+			return grpVersionSchema
+		}
+	}
+	// Query mapper
+	if val, ok := getValue(kind); ok {
+		return *val
+	}
+	utilruntime.HandleError(fmt.Errorf("Resouce '%s' not registered", kind))
+	return grpVersionSchema
+}
+
+func getValue(kind string) (*schema.GroupVersionResource, bool) {
+	if val, ok := groupVersionMapper[kind]; ok {
+		return &val, true
+	}
+	return nil, false
+}
+
+func refreshRegisteredResources(mapper map[string]schema.GroupVersionResource, clientConfig *rest.Config) error {
+	// build kubernetes client
+	client, err := newKubeClient(clientConfig)
+	if err != nil {
+		return err
+	}
+	// get registered server groups and resources
+	_, resourceList, err := client.Discovery().ServerGroupsAndResources()
+	if err != nil {
+		return err
+	}
+
+	for _, apiResource := range resourceList {
+		for _, resource := range apiResource.APIResources {
+			//			fmt.Println(resource.Name + " : " + apiResource.APIVersion + " , " + apiResource.GroupVersion + " , " + resource.Name)
+			grpVersion := strings.Split(apiResource.GroupVersion, "/")
+			if len(grpVersion) == 2 {
+				mapper[resource.Name] = schema.GroupVersionResource{
+					Group:    grpVersion[0],
+					Version:  grpVersion[1],
+					Resource: resource.Name,
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func newKubeClient(clientConfig *rest.Config) (*kubernetes.Clientset, error) {
+	client, err := kubernetes.NewForConfig(clientConfig)
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}

init.go

@@ -5,8 +5,8 @@ import (
 	"log"
 	"net/url"
 
+	client "github.com/nirmata/kube-policy/client"
 	"github.com/nirmata/kube-policy/config"
-	"github.com/nirmata/kube-policy/kubeclient"
 	"github.com/nirmata/kube-policy/utils"
 
 	rest "k8s.io/client-go/rest"
@@ -23,7 +23,7 @@ func createClientConfig(kubeconfig string) (*rest.Config, error) {
 	}
 }
 
-func initTlsPemPair(certFile, keyFile string, clientConfig *rest.Config, kubeclient *kubeclient.KubeClient) (*utils.TlsPemPair, error) {
+func initTlsPemPair(certFile, keyFile string, clientConfig *rest.Config, client *client.Client) (*utils.TlsPemPair, error) {
 	var tlsPair *utils.TlsPemPair
 	if certFile != "" || keyFile != "" {
 		tlsPair = tlsPairFromFiles(certFile, keyFile)
@@ -34,7 +34,7 @@ func initTlsPemPair(certFile, keyFile string, clientConfig *rest.Config, kubecli
 		log.Print("Using given TLS key/certificate pair")
 		return tlsPair, nil
 	} else {
-		tlsPair, err = tlsPairFromCluster(clientConfig, kubeclient)
+		tlsPair, err = tlsPairFromCluster(clientConfig, client)
 		if err == nil {
 			log.Printf("Using TLS key/certificate from cluster")
 		}
@@ -69,7 +69,7 @@ func tlsPairFromFiles(certFile, keyFile string) *utils.TlsPemPair {
 // Loads or creates PEM private key and TLS certificate for webhook server.
 // Created pair is stored in cluster's secret.
 // Returns struct with key/certificate pair.
-func tlsPairFromCluster(configuration *rest.Config, client *kubeclient.KubeClient) (*utils.TlsPemPair, error) {
+func tlsPairFromCluster(configuration *rest.Config, client *client.Client) (*utils.TlsPemPair, error) {
 	apiServerUrl, err := url.Parse(configuration.Host)
 	if err != nil {
 		return nil, err
@@ -79,7 +79,6 @@ func tlsPairFromCluster(configuration *rest.Config, client *kubeclient.KubeClien
 		Namespace:     config.KubePolicyNamespace,
 		ApiServerHost: apiServerUrl.Hostname(),
 	}
-
 	tlsPair := client.ReadTlsPair(certProps)
 	if utils.IsTlsPairShouldBeUpdated(tlsPair) {
 		log.Printf("Generating new key/certificate pair for TLS")

kubeclient/kubeclient.go

@@ -48,13 +48,8 @@ func (kc *KubeClient) GetEventsInterface(namespace string) event.EventInterface
 }
 
 func (kc *KubeClient) GetKubePolicyDeployment() (*apps.Deployment, error) {
-	kubePolicyDeployment, err := kc.client.
+	kubePolicyDeployment, err := kc.client.AppsV1().Deployments(config.KubePolicyNamespace).
-		Apps().
+		Get(config.KubePolicyDeploymentName, meta.GetOptions{ResourceVersion: "1"})
-		Deployments(config.KubePolicyNamespace).
-		Get(config.KubePolicyDeploymentName, meta.GetOptions{
-			ResourceVersion:      "1",
-			IncludeUninitialized: false,
-		})
 
 	if err != nil {
 		return nil, err
@@ -135,8 +130,7 @@ func (kc *KubeClient) GenerateSecret(generator types.PolicyConfigGenerator, name
 
 func defaultGetOptions() metav1.GetOptions {
 	return metav1.GetOptions{
-		ResourceVersion:      "1",
+		ResourceVersion: "1",
-		IncludeUninitialized: true,
 	}
 }
 

main.go

@@ -4,16 +4,15 @@ import (
 	"flag"
 	"log"
 
-	"github.com/nirmata/kube-policy/kubeclient"
 	"github.com/nirmata/kube-policy/policycontroller"
 	"github.com/nirmata/kube-policy/server"
 	"github.com/nirmata/kube-policy/webhooks"
 
+	client "github.com/nirmata/kube-policy/client"
 	policyclientset "github.com/nirmata/kube-policy/pkg/client/clientset/versioned"
 	informers "github.com/nirmata/kube-policy/pkg/client/informers/externalversions"
-	policyviolation "github.com/nirmata/kube-policy/pkg/policyviolation"
-
 	event "github.com/nirmata/kube-policy/pkg/event"
+	policyviolation "github.com/nirmata/kube-policy/pkg/policyviolation"
 	"k8s.io/sample-controller/pkg/signals"
 )
 
@@ -29,9 +28,9 @@ func main() {
 		log.Fatalf("Error building kubeconfig: %v\n", err)
 	}
 
-	kubeclient, err := kubeclient.NewKubeClient(clientConfig, nil)
+	client, err := client.NewDynamicClient(clientConfig, nil)
 	if err != nil {
-		log.Fatalf("Error creating kubeclient: %v\n", err)
+		log.Fatalf("Error creating client: %v\n", err)
 	}
 
 	policyClientset, err := policyclientset.NewForConfig(clientConfig)
@@ -43,18 +42,18 @@ func main() {
 	policyInformerFactory := informers.NewSharedInformerFactory(policyClientset, 0)
 	policyInformer := policyInformerFactory.Nirmata().V1alpha1().Policies()
 
-	eventController := event.NewEventController(kubeclient, policyInformer.Lister(), nil)
+	eventController := event.NewEventController(client, policyInformer.Lister(), nil)
-	violationBuilder := policyviolation.NewPolicyViolationBuilder(kubeclient, policyInformer.Lister(), policyClientset, eventController, nil)
+	violationBuilder := policyviolation.NewPolicyViolationBuilder(client, policyInformer.Lister(), policyClientset, eventController, nil)
 
 	policyController := policycontroller.NewPolicyController(policyClientset,
+		client,
 		policyInformer,
 		violationBuilder,
 		eventController,
-		nil,
+		nil)
-		kubeclient)
 
 	mutationWebhook, err := webhooks.CreateMutationWebhook(clientConfig,
-		kubeclient,
+		client,
 		policyInformer.Lister(),
 		eventController,
 		nil)
@@ -62,7 +61,7 @@ func main() {
 		log.Fatalf("Error creating mutation webhook: %v\n", err)
 	}
 
-	tlsPair, err := initTlsPemPair(cert, key, clientConfig, kubeclient)
+	tlsPair, err := initTlsPemPair(cert, key, clientConfig, client)
 	if err != nil {
 		log.Fatalf("Failed to initialize TLS key/certificate pair: %v\n", err)
 	}
@@ -85,6 +84,7 @@ func main() {
 
 	<-stopCh
 	server.Stop()
+	policyController.Stop()
 }
 
 func init() {

pkg/event/eventcontroller.go

@@ -3,9 +3,10 @@ package event
 import (
 	"fmt"
 	"log"
+	"os"
 	"time"
 
-	kubeClient "github.com/nirmata/kube-policy/kubeclient"
+	client "github.com/nirmata/kube-policy/client"
 	"github.com/nirmata/kube-policy/pkg/client/clientset/versioned/scheme"
 	policyscheme "github.com/nirmata/kube-policy/pkg/client/clientset/versioned/scheme"
 	policylister "github.com/nirmata/kube-policy/pkg/client/listers/policy/v1alpha1"
@@ -20,7 +21,7 @@ import (
 )
 
 type controller struct {
-	kubeClient   *kubeClient.KubeClient
+	client       *client.Client
 	policyLister policylister.PolicyLister
 	queue        workqueue.RateLimitingInterface
 	recorder     record.EventRecorder
@@ -36,30 +37,41 @@ type Generator interface {
 type Controller interface {
 	Generator
 	Run(stopCh <-chan struct{}) error
+	Stop()
 }
 
 //NewEventController to generate a new event controller
-func NewEventController(kubeClient *kubeClient.KubeClient,
+func NewEventController(client *client.Client,
 	policyLister policylister.PolicyLister,
 	logger *log.Logger) Controller {
+
+	if logger == nil {
+		logger = log.New(os.Stdout, "Event Controller: ", log.LstdFlags)
+	}
+
 	controller := &controller{
-		kubeClient:   kubeClient,
+		client:       client,
 		policyLister: policyLister,
 		queue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), eventWorkQueueName),
-		recorder:     initRecorder(kubeClient),
+		recorder:     initRecorder(client),
 		logger:       logger,
 	}
 	return controller
 }
 
-func initRecorder(kubeClient *kubeClient.KubeClient) record.EventRecorder {
+func initRecorder(client *client.Client) record.EventRecorder {
 	// Initliaze Event Broadcaster
 	policyscheme.AddToScheme(scheme.Scheme)
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(log.Printf)
+	eventInterface, err := client.GetEventsInterface()
+	if err != nil {
+		utilruntime.HandleError(err) // TODO: add more specific error
+		return nil
+	}
 	eventBroadcaster.StartRecordingToSink(
 		&typedcorev1.EventSinkImpl{
-			Interface: kubeClient.GetEventsInterface("")})
+			Interface: eventInterface})
 	recorder := eventBroadcaster.NewRecorder(
 		scheme.Scheme,
 		v1.EventSource{Component: eventSource})
@@ -81,11 +93,12 @@ func (c *controller) Run(stopCh <-chan struct{}) error {
 		go wait.Until(c.runWorker, time.Second, stopCh)
 	}
 	log.Println("Started eventbuilder controller workers")
-	<-stopCh
-	log.Println("Shutting down eventbuilder controller workers")
 	return nil
 }
 
+func (c *controller) Stop() {
+	log.Println("Shutting down eventbuilder controller workers")
+}
 func (c *controller) runWorker() {
 	for c.processNextWorkItem() {
 	}
@@ -122,20 +135,32 @@ func (c *controller) processNextWorkItem() bool {
 func (c *controller) SyncHandler(key Info) error {
 	var resource runtime.Object
 	var err error
+	namespace, name, err := cache.SplitMetaNamespaceKey(key.Resource)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key.Resource))
+		return err
+	}
+
 	switch key.Kind {
 	case "Policy":
-		namespace, name, err := cache.SplitMetaNamespaceKey(key.Resource)
+		//TODO: policy is clustered resource so wont need namespace
-		if err != nil {
-			utilruntime.HandleError(fmt.Errorf("unable to extract namespace and name for %s", key.Resource))
-			return err
-		}
 		resource, err = c.policyLister.Policies(namespace).Get(name)
 		if err != nil {
 			utilruntime.HandleError(fmt.Errorf("unable to create event for policy %s, will retry ", key.Resource))
 			return err
 		}
 	default:
-		resource, err = c.kubeClient.GetResource(key.Kind, key.Resource)
+		resource, err = c.client.GetResource(key.Kind, namespace, name)
+		if err != nil {
+			return err
+		}
+		//TODO: Test if conversion from unstructured to runtime.Object is implicit or explicit conversion is required
+		// resourceobj, err := client.ConvertToRuntimeObject(resource)
+		// if err != nil {
+		// 	return err
+		// }
+
+		//		resource, err = c.kubeClient.GetResource(key.Kind, key.Resource)
 		if err != nil {
 			utilruntime.HandleError(fmt.Errorf("unable to create event for resource %s, will retry ", key.Resource))
 			return err

pkg/policyviolation/builder.go

@@ -3,8 +3,9 @@ package policyviolation
 import (
 	"fmt"
 	"log"
+	"os"
 
-	kubeClient "github.com/nirmata/kube-policy/kubeclient"
+	client "github.com/nirmata/kube-policy/client"
 	types "github.com/nirmata/kube-policy/pkg/apis/policy/v1alpha1"
 	policyclientset "github.com/nirmata/kube-policy/pkg/client/clientset/versioned"
 	policylister "github.com/nirmata/kube-policy/pkg/client/listers/policy/v1alpha1"
@@ -19,7 +20,7 @@ type Generator interface {
 }
 
 type builder struct {
-	kubeClient      *kubeClient.KubeClient
+	client          *client.Client
 	policyLister    policylister.PolicyLister
 	policyInterface policyclientset.Interface
 	eventBuilder    event.Generator
@@ -34,15 +35,18 @@ type Builder interface {
 }
 
 //NewPolicyViolationBuilder returns new violation builder
-func NewPolicyViolationBuilder(
+func NewPolicyViolationBuilder(client *client.Client,
-	kubeClient *kubeClient.KubeClient,
 	policyLister policylister.PolicyLister,
 	policyInterface policyclientset.Interface,
 	eventController event.Generator,
 	logger *log.Logger) Builder {
 
+	if logger == nil {
+		logger = log.New(os.Stdout, "Violation Builder: ", log.LstdFlags)
+	}
+
 	builder := &builder{
-		kubeClient:      kubeClient,
+		client:          client,
 		policyLister:    policyLister,
 		policyInterface: policyInterface,
 		eventBuilder:    eventController,
@@ -97,8 +101,15 @@ func (b *builder) processViolation(info Info) error {
 }
 
 func (b *builder) isActive(kind string, resource string) (bool, error) {
+	namespace, name, err := cache.SplitMetaNamespaceKey(resource)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", resource))
+		return false, err
+	}
 	// Generate Merge Patch
-	_, err := b.kubeClient.GetResource(kind, resource)
+	//TODO: test for namspace and clustered object
+	_, err = b.client.GetResource(kind, namespace, name)
+	//	_, err := b.kubeClient.GetResource(kind, resource)
 	if err != nil {
 		utilruntime.HandleError(fmt.Errorf("unable to get resource %s ", resource))
 		return false, err

policycontroller/policycontroller.go

@@ -3,9 +3,10 @@ package policycontroller
 import (
 	"fmt"
 	"log"
+	"os"
 	"time"
 
-	kubeClient "github.com/nirmata/kube-policy/kubeclient"
+	client "github.com/nirmata/kube-policy/client"
 	types "github.com/nirmata/kube-policy/pkg/apis/policy/v1alpha1"
 	policyclientset "github.com/nirmata/kube-policy/pkg/client/clientset/versioned"
 	infomertypes "github.com/nirmata/kube-policy/pkg/client/informers/externalversions/policy/v1alpha1"
@@ -23,7 +24,7 @@ import (
 
 //PolicyController to manage Policy CRD
 type PolicyController struct {
-	kubeClient       *kubeClient.KubeClient
+	client           *client.Client
 	policyLister     lister.PolicyLister
 	policyInterface  policyclientset.Interface
 	policySynced     cache.InformerSynced
@@ -35,14 +36,17 @@ type PolicyController struct {
 
 // NewPolicyController from cmd args
 func NewPolicyController(policyInterface policyclientset.Interface,
+	client *client.Client,
 	policyInformer infomertypes.PolicyInformer,
 	violationBuilder policyviolation.Generator,
 	eventController event.Generator,
-	logger *log.Logger,
+	logger *log.Logger) *PolicyController {
-	kubeClient *kubeClient.KubeClient) *PolicyController {
 
+	if logger == nil {
+		logger = log.New(os.Stdout, "Policy Controller: ", log.LstdFlags)
+	}
 	controller := &PolicyController{
-		kubeClient:       kubeClient,
+		client:           client,
 		policyLister:     policyInformer.Lister(),
 		policyInterface:  policyInterface,
 		policySynced:     policyInformer.Informer().HasSynced,
@@ -86,6 +90,7 @@ func (pc *PolicyController) deletePolicyHandler(resource interface{}) {
 func (pc *PolicyController) enqueuePolicy(obj interface{}) {
 	var key string
 	var err error
+	pc.logger.Println("enque")
 	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
 		utilruntime.HandleError(err)
 		return
@@ -109,13 +114,14 @@ func (pc *PolicyController) Run(stopCh <-chan struct{}) error {
 	for i := 0; i < policyControllerWorkerCount; i++ {
 		go wait.Until(pc.runWorker, time.Second, stopCh)
 	}
-
 	pc.logger.Println("started policy controller workers")
-	<-stopCh
+
-	pc.logger.Println("shutting down policy controller workers")
 	return nil
 }
 
+func (pc *PolicyController) Stop() {
+	pc.logger.Println("shutting down policy controller workers")
+}
 func (pc *PolicyController) runWorker() {
 	for pc.processNextWorkItem() {
 	}

policycontroller/processPolicy.go

@@ -1,7 +1,6 @@
 package policycontroller
 
 import (
-	"encoding/json"
 	"errors"
 	"fmt"
 
@@ -11,7 +10,6 @@ import (
 	"github.com/nirmata/kube-policy/pkg/policyengine/mutation"
 	"github.com/nirmata/kube-policy/pkg/policyviolation"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 )
 
@@ -24,7 +22,7 @@ func (pc *PolicyController) runForPolicy(key string) {
 	}
 
 	if policy == nil {
-		pc.logger.Printf("Counld not find policy by key %s", key)
+		pc.logger.Printf("Could not find policy by key %s", key)
 		return
 	}
 
@@ -53,13 +51,12 @@ func (pc *PolicyController) processPolicy(policy types.Policy) (
 		}
 
 		for _, resource := range resources {
-			rawResource, err := json.Marshal(resource)
 			if err != nil {
 				pc.logger.Printf("Failed to marshal resources map to rule %s, err: %v\n", rule.Name, err)
 				continue
 			}
 
-			violation, eventInfos, err := pc.processExisting(policy, rawResource)
+			violation, eventInfos, err := pc.processExisting(policy, resource)
 			if err != nil {
 				pc.logger.Printf("Failed to process rule %s, err: %v\n", rule.Name, err)
 				continue
@@ -72,8 +69,8 @@ func (pc *PolicyController) processPolicy(policy types.Policy) (
 	return violations, events, nil
 }
 
-func (pc *PolicyController) filterResourceByRule(rule types.PolicyRule) ([]runtime.Object, error) {
+func (pc *PolicyController) filterResourceByRule(rule types.PolicyRule) ([][]byte, error) {
-	var targetResources []runtime.Object
+	var targetResources [][]byte
 	// TODO: make this namespace all
 	var namespace = "default"
 	if err := rule.Validate(); err != nil {
@@ -81,18 +78,18 @@ func (pc *PolicyController) filterResourceByRule(rule types.PolicyRule) ([]runti
 	}
 
 	// Get the resource list from kind
-	resources, err := pc.kubeClient.ListResource(rule.Resource.Kind, namespace)
+	resources, err := pc.client.ListResource(rule.Resource.Kind, namespace)
+	//	resources, err := pc.kubeClient.ListResource(rule.Resource.Kind, namespace)
 	if err != nil {
 		return nil, err
 	}
 
-	for _, resource := range resources {
+	for _, resource := range resources.Items {
 		// TODO:
-		rawResource, err := json.Marshal(resource)
+		rawResource, err := resource.MarshalJSON()
 		// objKind := resource.GetObjectKind()
 		// codecFactory := serializer.NewCodecFactory(runtime.NewScheme())
 		// codecFactory.EncoderForVersion()
-
 		if err != nil {
 			pc.logger.Printf("failed to marshal object %v", resource)
 			continue
@@ -100,7 +97,7 @@ func (pc *PolicyController) filterResourceByRule(rule types.PolicyRule) ([]runti
 
 		// filter the resource by name and label
 		if ok, _ := mutation.IsRuleApplicableToResource(rawResource, rule.Resource); ok {
-			targetResources = append(targetResources, resource)
+			targetResources = append(targetResources, rawResource)
 		}
 	}
 	return targetResources, nil
@@ -162,12 +159,12 @@ func (pc *PolicyController) applyGenerate(generatedDataList []policyengine.Gener
 	for _, generateData := range generatedDataList {
 		switch generateData.ConfigKind {
 		case "ConfigMap":
-			err := pc.kubeClient.GenerateConfigMap(generateData.Generator, generateData.Namespace)
+			err := pc.client.GenerateConfigMap(generateData.Generator, generateData.Namespace)
 			if err != nil {
 				return err
 			}
 		case "Secret":
-			err := pc.kubeClient.GenerateSecret(generateData.Generator, generateData.Namespace)
+			err := pc.client.GenerateSecret(generateData.Generator, generateData.Namespace)
 			if err != nil {
 				return err
 			}

webhooks/mutation.go

@@ -7,7 +7,7 @@ import (
 	"os"
 	"sort"
 
-	kubeclient "github.com/nirmata/kube-policy/kubeclient"
+	client "github.com/nirmata/kube-policy/client"
 	types "github.com/nirmata/kube-policy/pkg/apis/policy/v1alpha1"
 	policylister "github.com/nirmata/kube-policy/pkg/client/listers/policy/v1alpha1"
 	event "github.com/nirmata/kube-policy/pkg/event"
@@ -24,7 +24,7 @@ import (
 // MutationWebhook is a data type that represents
 // business logic for resource mutation
 type MutationWebhook struct {
-	kubeclient   *kubeclient.KubeClient
+	client       *client.Client
 	policyLister policylister.PolicyLister
 	registration *MutationWebhookRegistration
 	eventBuilder event.Generator
@@ -34,15 +34,15 @@ type MutationWebhook struct {
 // Registers mutation webhook in cluster and creates object for this webhook
 func CreateMutationWebhook(
 	clientConfig *rest.Config,
-	kubeclient *kubeclient.KubeClient,
+	client *client.Client,
 	policyLister policylister.PolicyLister,
 	eventController event.Generator,
 	logger *log.Logger) (*MutationWebhook, error) {
-	if clientConfig == nil || kubeclient == nil {
+	if clientConfig == nil || client == nil {
 		return nil, errors.New("Some parameters are not set")
 	}
 
-	registration, err := NewMutationWebhookRegistration(clientConfig, kubeclient)
+	registration, err := NewMutationWebhookRegistration(clientConfig, client)
 	if err != nil {
 		return nil, err
 	}
@@ -57,7 +57,7 @@ func CreateMutationWebhook(
 	}
 
 	return &MutationWebhook{
-		kubeclient:   kubeclient,
+		client:       client,
 		policyLister: policyLister,
 		registration: registration,
 		eventBuilder: eventController,
@@ -207,9 +207,9 @@ func (mw *MutationWebhook) applyConfigGenerator(generator *types.PolicyConfigGen
 
 	switch configKind {
 	case "ConfigMap":
-		err = mw.kubeclient.GenerateConfigMap(*generator, namespace)
+		err = mw.client.GenerateConfigMap(*generator, namespace)
 	case "Secret":
-		err = mw.kubeclient.GenerateSecret(*generator, namespace)
+		err = mw.client.GenerateSecret(*generator, namespace)
 	default:
 		err = errors.New(fmt.Sprintf("Unsupported config Kind '%s'", configKind))
 	}

webhooks/registration.go

@@ -4,8 +4,8 @@ import (
 	"errors"
 	"io/ioutil"
 
+	client "github.com/nirmata/kube-policy/client"
 	"github.com/nirmata/kube-policy/config"
-	kubeclient "github.com/nirmata/kube-policy/kubeclient"
 
 	admregapi "k8s.io/api/admissionregistration/v1beta1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -15,11 +15,12 @@ import (
 
 type MutationWebhookRegistration struct {
 	registrationClient *admregclient.AdmissionregistrationV1beta1Client
-	kubeclient         *kubeclient.KubeClient
+	client             *client.Client
 	clientConfig       *rest.Config
 }
 
-func NewMutationWebhookRegistration(clientConfig *rest.Config, kubeclient *kubeclient.KubeClient) (*MutationWebhookRegistration, error) {
+func NewMutationWebhookRegistration(clientConfig *rest.Config,
+	client *client.Client) (*MutationWebhookRegistration, error) {
 	registrationClient, err := admregclient.NewForConfig(clientConfig)
 	if err != nil {
 		return nil, err
@@ -27,7 +28,7 @@ func NewMutationWebhookRegistration(clientConfig *rest.Config, kubeclient *kubec
 
 	return &MutationWebhookRegistration{
 		registrationClient: registrationClient,
-		kubeclient:         kubeclient,
+		client:             client,
 		clientConfig:       clientConfig,
 	}, nil
 }
@@ -56,7 +57,7 @@ func (mwr *MutationWebhookRegistration) constructWebhookConfig(configuration *re
 		return nil, errors.New("Unable to extract CA data from configuration")
 	}
 
-	kubePolicyDeployment, err := mwr.kubeclient.GetKubePolicyDeployment()
+	kubePolicyDeployment, err := mwr.client.GetKubePolicyDeployment()
 
 	if err != nil {
 		return nil, err