Feature: Stream changes to gui

cmd/capacitor/main.go

@@ -57,6 +57,14 @@ func main() {
 	runController(err, helmReleaseController, stopCh)
 	eventController, err := controllers.EventController(client, dynamicClient, clientHub)
 	runController(err, eventController, stopCh)
+	deploymentController, err := controllers.DeploymentController(client, dynamicClient, clientHub)
+	runController(err, deploymentController, stopCh)
+	podController, err := controllers.PodController(client, dynamicClient, clientHub)
+	runController(err, podController, stopCh)
+	serviceController, err := controllers.ServiceController(client, dynamicClient, clientHub)
+	runController(err, serviceController, stopCh)
+	ingressController, err := controllers.IngressController(client, dynamicClient, clientHub)
+	runController(err, ingressController, stopCh)
 
 	r := api.SetupRouter(client, dynamicClient, config, clientHub, runningLogStreams)
 	go func() {

pkg/controllers/deploymentController.go

@@ -0,0 +1,50 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	apps_v1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+func DeploymentController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	deploymentListWatcher := cache.NewListWatchFromClient(client.AppsV1().RESTClient(), "deployments", v1.NamespaceAll, fields.Everything())
+	deploymentController := NewController(
+		"deployment",
+		deploymentListWatcher,
+		&apps_v1.Deployment{},
+		func(informerEvent Event, objectMeta meta_v1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				services, err := flux.Services(client, dynamicClient)
+				if err != nil {
+					panic(err.Error())
+				}
+				servicesBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.SERVICES_RECEIVED,
+					Payload: services,
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+				clientHub.Broadcast <- servicesBytes
+			}
+			return nil
+		})
+	return deploymentController, nil
+}

pkg/controllers/ingressController.go

@@ -0,0 +1,50 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	v1 "k8s.io/api/core/v1"
+	networking_v1 "k8s.io/api/networking/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+func IngressController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	ingressListWatcher := cache.NewListWatchFromClient(client.NetworkingV1().RESTClient(), "ingresses", v1.NamespaceAll, fields.Everything())
+	ingressController := NewController(
+		"ingress",
+		ingressListWatcher,
+		&networking_v1.Ingress{},
+		func(informerEvent Event, objectMeta meta_v1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				services, err := flux.Services(client, dynamicClient)
+				if err != nil {
+					panic(err.Error())
+				}
+				servicesBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.SERVICES_RECEIVED,
+					Payload: services,
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+				clientHub.Broadcast <- servicesBytes
+			}
+			return nil
+		})
+	return ingressController, nil
+}

pkg/controllers/podController.go

@@ -0,0 +1,49 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	v1 "k8s.io/api/core/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+func PodController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	podListWatcher := cache.NewListWatchFromClient(client.CoreV1().RESTClient(), "pods", v1.NamespaceAll, fields.Everything())
+	podController := NewController(
+		"pod",
+		podListWatcher,
+		&v1.Pod{},
+		func(informerEvent Event, objectMeta meta_v1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				services, err := flux.Services(client, dynamicClient)
+				if err != nil {
+					panic(err.Error())
+				}
+				servicesBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.SERVICES_RECEIVED,
+					Payload: services,
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+				clientHub.Broadcast <- servicesBytes
+			}
+			return nil
+		})
+	return podController, nil
+}

pkg/controllers/serviceController.go

@@ -0,0 +1,49 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	v1 "k8s.io/api/core/v1"
+	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+func ServiceController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	serviceListWatcher := cache.NewListWatchFromClient(client.CoreV1().RESTClient(), "services", v1.NamespaceAll, fields.Everything())
+	serviceController := NewController(
+		"service",
+		serviceListWatcher,
+		&v1.Service{},
+		func(informerEvent Event, objectMeta meta_v1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				services, err := flux.Services(client, dynamicClient)
+				if err != nil {
+					panic(err.Error())
+				}
+				servicesBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.SERVICES_RECEIVED,
+					Payload: services,
+				})
+				if err != nil {
+					panic(err.Error())
+				}
+				clientHub.Broadcast <- servicesBytes
+			}
+			return nil
+		})
+	return serviceController, nil
+}

pkg/streaming/types.go

@@ -3,6 +3,7 @@ package streaming
 const (
 	FLUX_STATE_RECEIVED  string = "FLUX_STATE_RECEIVED"
 	FLUX_EVENTS_RECEIVED string = "FLUX_EVENTS_RECEIVED"
+	SERVICES_RECEIVED    string = "SERVICES_RECEIVED"
 	POD_LOGS_RECEIVED    string = "POD_LOGS_RECEIVED"
 )