Merge pull request #60 from bonifaido/extensions

Extensions

Dockerfile.build

@@ -4,8 +4,11 @@ MAINTAINER "Cuong Manh Le <cuong.manhle.vn@gmail.com>"
 RUN apk update && \
     apk add git build-base && \
     rm -rf /var/cache/apk/* && \
-    go get -v -d github.com/skippbox/kubewatch && \
-    cd "$GOPATH/src/github.com/skippbox/kubewatch" && \
+    mkdir -p "$GOPATH/src/github.com/skippbox/kubewatch"
+
+ADD . "$GOPATH/src/github.com/skippbox/kubewatch"
+
+RUN cd "$GOPATH/src/github.com/skippbox/kubewatch" && \
     CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -a --installsuffix cgo --ldflags="-s" -o /kubewatch
 
 COPY Dockerfile.run /

config/config.go

@@ -17,10 +17,10 @@ limitations under the License.
 package config
 
 import (
-	"os"
-	"runtime"
 	"io/ioutil"
+	"os"
 	"path/filepath"
+	"runtime"
 
 	"gopkg.in/yaml.v2"
 )
@@ -39,11 +39,13 @@ type Resource struct {
 	DaemonSet             bool `json:"ds"`
 	Services              bool `json:"svc"`
 	Pod                   bool `json:"po"`
+	Job                   bool `json:"job"`
+	PersistentVolume      bool `json:"pv"`
 }
 
 // Config struct contains kubewatch configuration
 type Config struct {
-	Handler  Handler  `json:"handler"`
+	Handler Handler `json:"handler"`
 	//Reason   []string `json:"reason"`
 	Resource Resource `json:"resource"`
 }
@@ -125,6 +127,12 @@ func (c *Config) CheckMissingResourceEnvvars() {
 	if !c.Resource.Services && os.Getenv("KW_SERVICE") == "true" {
 		c.Resource.Services = true
 	}
+	if !c.Resource.Job && os.Getenv("KW_JOB") == "true" {
+		c.Resource.Job = true
+	}
+	if !c.Resource.PersistentVolume && os.Getenv("KW_PERSISTENT_VOLUME") == "true" {
+		c.Resource.PersistentVolume = true
+	}
 	if (c.Handler.Slack.Channel == "") && (os.Getenv("SLACK_CHANNEL") != "") {
 		c.Handler.Slack.Channel = os.Getenv("SLACK_CHANNEL")
 	}

pkg/controller/controller.go

@@ -24,6 +24,7 @@ import (
 	"github.com/skippbox/kubewatch/config"
 	"github.com/skippbox/kubewatch/pkg/handlers"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 	"k8s.io/kubernetes/pkg/client/cache"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/controller/framework"
@@ -41,26 +42,37 @@ func Controller(conf *config.Config, eventHandler handlers.Handler) {
 	}
 
 	kubeClient := client.NewOrDie(kubeConfig)
+	kubeExtensionsClient := client.NewExtensionsOrDie(kubeConfig)
 
 	if conf.Resource.Pod {
-		var podsStore cache.Store
-		podsStore = watchPods(kubeClient, podsStore, eventHandler)
+		watchPods(kubeClient, eventHandler)
 	}
 
 	if conf.Resource.Services {
-		var servicesStore cache.Store
-		servicesStore = watchServices(kubeClient, servicesStore, eventHandler)
+		watchServices(kubeClient, eventHandler)
 	}
 
 	if conf.Resource.ReplicationController {
-		var rcStore cache.Store
-		rcStore = watchReplicationControllers(kubeClient, rcStore, eventHandler)
+		watchReplicationControllers(kubeClient, eventHandler)
+	}
+
+	if conf.Resource.Deployment {
+		watchDeployments(kubeExtensionsClient, eventHandler)
+	}
+
+	if conf.Resource.Job {
+		watchJobs(kubeExtensionsClient, eventHandler)
+	}
+
+	if conf.Resource.PersistentVolume {
+		var servicesStore cache.Store
+		servicesStore = watchPersistenVolumes(kubeClient, servicesStore, eventHandler)
 	}
 
 	logrus.Fatal(http.ListenAndServe(":8081", nil))
 }
 
-func watchPods(client *client.Client, store cache.Store, eventHandler handlers.Handler) cache.Store {
+func watchPods(client *client.Client, eventHandler handlers.Handler) cache.Store {
 	//Define what we want to look for (Pods)
 	watchlist := cache.NewListWatchFromClient(client, "pods", api.NamespaceAll, fields.Everything())
 
@@ -83,7 +95,7 @@ func watchPods(client *client.Client, store cache.Store, eventHandler handlers.H
 	return eStore
 }
 
-func watchServices(client *client.Client, store cache.Store, eventHandler handlers.Handler) cache.Store {
+func watchServices(client *client.Client, eventHandler handlers.Handler) cache.Store {
 	//Define what we want to look for (Services)
 	watchlist := cache.NewListWatchFromClient(client, "services", api.NamespaceAll, fields.Everything())
 
@@ -107,7 +119,7 @@ func watchServices(client *client.Client, store cache.Store, eventHandler handle
 	return eStore
 }
 
-func watchReplicationControllers(client *client.Client, store cache.Store, eventHandler handlers.Handler) cache.Store {
+func watchReplicationControllers(client *client.Client, eventHandler handlers.Handler) cache.Store {
 	//Define what we want to look for (ReplicationControllers)
 	watchlist := cache.NewListWatchFromClient(client, "replicationcontrollers", api.NamespaceAll, fields.Everything())
 
@@ -129,3 +141,72 @@ func watchReplicationControllers(client *client.Client, store cache.Store, event
 
 	return eStore
 }
+
+func watchDeployments(client *client.ExtensionsClient, eventHandler handlers.Handler) cache.Store {
+	//Define what we want to look for (Deployments)
+	watchlist := cache.NewListWatchFromClient(client, "deployments", api.NamespaceAll, fields.Everything())
+
+	resyncPeriod := 30 * time.Minute
+
+	//Setup an informer to call functions when the watchlist changes
+	eStore, eController := framework.NewInformer(
+		watchlist,
+		&v1beta1.Deployment{},
+		resyncPeriod,
+		framework.ResourceEventHandlerFuncs{
+			AddFunc:    eventHandler.ObjectCreated,
+			DeleteFunc: eventHandler.ObjectDeleted,
+		},
+	)
+
+	//Run the controller as a goroutine
+	go eController.Run(wait.NeverStop)
+
+	return eStore
+}
+
+func watchJobs(client *client.ExtensionsClient, eventHandler handlers.Handler) cache.Store {
+	//Define what we want to look for (Jobs)
+	watchlist := cache.NewListWatchFromClient(client, "jobs", api.NamespaceAll, fields.Everything())
+
+	resyncPeriod := 30 * time.Minute
+
+	//Setup an informer to call functions when the watchlist changes
+	eStore, eController := framework.NewInformer(
+		watchlist,
+		&v1beta1.Job{},
+		resyncPeriod,
+		framework.ResourceEventHandlerFuncs{
+			AddFunc:    eventHandler.ObjectCreated,
+			DeleteFunc: eventHandler.ObjectDeleted,
+		},
+	)
+
+	//Run the controller as a goroutine
+	go eController.Run(wait.NeverStop)
+
+	return eStore
+}
+
+func watchPersistenVolumes(client *client.Client, store cache.Store, eventHandler handlers.Handler) cache.Store {
+	//Define what we want to look for (PersistenVolumes)
+	watchlist := cache.NewListWatchFromClient(client, "persistentvolumes", api.NamespaceAll, fields.Everything())
+
+	resyncPeriod := 30 * time.Minute
+
+	//Setup an informer to call functions when the watchlist changes
+	eStore, eController := framework.NewInformer(
+		watchlist,
+		&api.PersistentVolume{},
+		resyncPeriod,
+		framework.ResourceEventHandlerFuncs{
+			AddFunc:    eventHandler.ObjectCreated,
+			DeleteFunc: eventHandler.ObjectDeleted,
+		},
+	)
+
+	//Run the controller as a goroutine
+	go eController.Run(wait.NeverStop)
+
+	return eStore
+}

pkg/event/event.go

@@ -18,6 +18,8 @@ package event
 
 import (
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apis/extensions"
+	"k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 )
 
 // Event represent an event got from k8s api server
@@ -44,24 +46,41 @@ func New(obj interface{}, action string) Event {
 	var namespace, kind, component, host, reason, status, name string
 	if apiService, ok := obj.(*api.Service); ok {
 		namespace = apiService.ObjectMeta.Namespace
+		name = apiService.Name
 		kind = "service"
 		component = string(apiService.Spec.Type)
 		reason = action
 		status = m[action]
-		name = apiService.Name
 	} else if apiPod, ok := obj.(*api.Pod); ok {
 		namespace = apiPod.ObjectMeta.Namespace
+		name = apiPod.Name
 		kind = "pod"
 		reason = action
 		host = apiPod.Spec.NodeName
 		status = m[action]
-		name = apiPod.Name
 	} else if apiRC, ok := obj.(*api.ReplicationController); ok {
-		name = apiRC.TypeMeta.Kind
+		namespace = apiRC.ObjectMeta.Namespace
+		name = apiRC.Name
 		kind = "replication controller"
 		reason = action
 		status = m[action]
-		name = apiRC.Name
+	} else if apiDeployment, ok := obj.(*extensions.Deployment); ok {
+		namespace = apiDeployment.ObjectMeta.Namespace
+		name = apiDeployment.Name
+		kind = "deployment"
+		reason = action
+		status = m[action]
+	} else if apiJob, ok := obj.(*v1beta1.Job); ok {
+		namespace = apiJob.ObjectMeta.Namespace
+		name = apiJob.Name
+		kind = "job"
+		reason = action
+		status = m[action]
+	} else if apiPV, ok := obj.(*api.PersistentVolume); ok {
+		name = apiPV.Name
+		kind = "persistent volume"
+		reason = action
+		status = m[action]
 	}
 
 	kbEvent := Event{