Minor fix

cmd/capacitor/main.go

@@ -51,6 +51,10 @@ func main() {
 	runController(err, ociRepositoryController, stopCh)
 	bucketController, err := controllers.BucketController(client, dynamicClient, clientHub)
 	runController(err, bucketController, stopCh)
+	helmRepositoryController, err := controllers.HelmRepositoryController(client, dynamicClient, clientHub)
+	runController(err, helmRepositoryController, stopCh)
+	helmChartController, err := controllers.HelmChartController(client, dynamicClient, clientHub)
+	runController(err, helmChartController, stopCh)
 	kustomizationController, err := controllers.KustomizeController(client, dynamicClient, clientHub)
 	runController(err, kustomizationController, stopCh)
 	helmReleaseController, err := controllers.HelmReleaseController(client, dynamicClient, clientHub)

deploy/k8s/rbac.yaml

@@ -34,6 +34,8 @@ rules:
     - gitrepositories
     - ocirepositories
     - buckets
+    - helmrepositories
+    - helmcharts
     - kustomizations
     - helmreleases
   verbs:

pkg/controllers/helmChartController.go

@@ -0,0 +1,54 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+)
+
+var helmChartResource = schema.GroupVersionResource{
+	Group:    "source.toolkit.fluxcd.io",
+	Version:  "v1beta2",
+	Resource: "helmcharts",
+}
+
+func HelmChartController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	return NewDynamicController(
+		"helmcharts.source.toolkit.fluxcd.io",
+		dynamicClient,
+		helmChartResource,
+		func(informerEvent Event, objectMeta metav1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				fluxState, err := flux.State(client, dynamicClient)
+				if err != nil {
+					logrus.Warnf("could not get flux state: %s", err)
+					return nil
+				}
+				fluxStateBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.FLUX_STATE_RECEIVED,
+					Payload: fluxState,
+				})
+				if err != nil {
+					logrus.Warnf("could not marshal event: %s", err)
+					return nil
+				}
+				clientHub.Broadcast <- fluxStateBytes
+			}
+			return nil
+		})
+}

pkg/controllers/helmRepositoryController.go

@@ -0,0 +1,54 @@
+package controllers
+
+import (
+	"encoding/json"
+
+	"github.com/gimlet-io/capacitor/pkg/flux"
+	"github.com/gimlet-io/capacitor/pkg/streaming"
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/kubernetes"
+)
+
+var helmRepositoryResource = schema.GroupVersionResource{
+	Group:    "source.toolkit.fluxcd.io",
+	Version:  "v1beta2",
+	Resource: "helmrepositories",
+}
+
+func HelmRepositoryController(
+	client *kubernetes.Clientset,
+	dynamicClient *dynamic.DynamicClient,
+	clientHub *streaming.ClientHub,
+) (*Controller, error) {
+	return NewDynamicController(
+		"helmrepositories.source.toolkit.fluxcd.io",
+		dynamicClient,
+		helmRepositoryResource,
+		func(informerEvent Event, objectMeta metav1.ObjectMeta, obj interface{}) error {
+			switch informerEvent.eventType {
+			case "create":
+				fallthrough
+			case "update":
+				fallthrough
+			case "delete":
+				fluxState, err := flux.State(client, dynamicClient)
+				if err != nil {
+					logrus.Warnf("could not get flux state: %s", err)
+					return nil
+				}
+				fluxStateBytes, err := json.Marshal(streaming.Envelope{
+					Type:    streaming.FLUX_STATE_RECEIVED,
+					Payload: fluxState,
+				})
+				if err != nil {
+					logrus.Warnf("could not marshal event: %s", err)
+					return nil
+				}
+				clientHub.Broadcast <- fluxStateBytes
+			}
+			return nil
+		})
+}