Merge pull request #33 from overmindtech/20-create-generic-kube-source

20 create generic Kube source

.devcontainer/setup.sh

@@ -1,3 +1,11 @@
 #!/bin/bash
 
-curl -Lo ./kind \"https://kind.sigs.k8s.io/dl/v0.14.0/kind-$(uname)-amd64\" && chmod +x ./kind && sudo mv ./kind /usr/local/bin/kind
+go install sigs.k8s.io/kind@latest
+
+# Create the test cluster (the tests also do this) but also set local kube
+# config
+kind create cluster --name k8s-source-tests
+kind export kubeconfig --name k8s-source-tests
+
+# Install k9s
+curl -sS https://webinstall.dev/k9s | bash

.github/workflows/test-build.yml

@@ -18,7 +18,7 @@ jobs:
         run: go vet ./...
 
       - name: Test
-        run: go test ./...
+        run: go test -v -race -timeout 3m ./...
 
   build:
     name: Build

.vscode/launch.json

@@ -0,0 +1,23 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Launch Package",
+            "type": "go",
+            "request": "launch",
+            "mode": "auto",
+            "program": "${workspaceFolder}/main.go",
+            "args": [
+                "--kubeconfig",
+                "/home/vscode/.kube/config",
+                "--log",
+                "debug",
+                "--nats-servers",
+                "nats://localhost:4222"
+            ]
+        }
+    ]
+}

.vscode/tasks.json

@@ -0,0 +1,12 @@
+{
+    // See https://go.microsoft.com/fwlink/?LinkId=733558
+    // for the documentation about the tasks.json format
+    "version": "2.0.0",
+    "tasks": [
+        {
+            "label": "Run NATS Server",
+            "type": "shell",
+            "command": "docker run  -p 4222:4222 -p 8222:8222 nats"
+        }
+    ]
+}

cmd/root.go

@@ -1,6 +1,8 @@
 package cmd
 
 import (
+	"context"
+	"crypto/sha1"
 	"errors"
 	"fmt"
 	"net/http"
@@ -16,9 +18,11 @@ import (
 	"github.com/nats-io/nkeys"
 	"github.com/overmindtech/connect"
 	"github.com/overmindtech/discovery"
-	"github.com/overmindtech/k8s-source/internal/sources"
+	"github.com/overmindtech/k8s-source/sources"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
@@ -32,17 +36,8 @@ var cfgFile string
 // rootCmd represents the base command when called without any subcommands
 var rootCmd = &cobra.Command{
 	Use:   "k8s-source",
-	Short: "Remote primary source for kubernetes",
-	Long: `This is designed to be run as part of srcman
-(https://github.com/overmindtech/srcman)
-
-It responds to requests for items relating to kubernetes clusters.
-Each namespace is a separate scope, as are non-namespaced resources
-within each cluster.
-
-This can be configured using a yaml file and the --config flag, or by
-using appropriately named environment variables, for example "nats-name-prefix"
-can be set using an environment variable named "NATS_NAME_PREFIX"
+	Short: "Kubernetes source",
+	Long: `Gathers details from existing kubernetes clusters
 `,
 	Run: func(cmd *cobra.Command, args []string) {
 		natsServers := viper.GetStringSlice("nats-servers")
@@ -108,8 +103,6 @@ can be set using an environment variable named "NATS_NAME_PREFIX"
 		// Now that we have a connection to the kubernetes cluster we need to go
 		// about generating some sources.
 		var k8sURL *url.URL
-		var nss sources.NamespaceStorage
-		var sourceList []discovery.Source
 
 		k8sURL, err = url.Parse(rc.Host)
 
@@ -131,32 +124,6 @@ can be set using an environment variable named "NATS_NAME_PREFIX"
 			}
 		}
 
-		sources.ClusterName = k8sURL.Host
-
-		// Get list of namspaces
-		nss = sources.NamespaceStorage{
-			CS:            clientSet,
-			CacheDuration: (10 * time.Second),
-		}
-
-		// Load all sources
-		for _, srcFunction := range sources.SourceFunctions {
-			src, err := srcFunction(clientSet)
-
-			if err != nil {
-				log.WithFields(log.Fields{
-					"error":      err,
-					"sourceName": src.Name(),
-				}).Error("Failed loading source")
-
-				continue
-			}
-
-			src.NSS = &nss
-
-			sourceList = append(sourceList, &src)
-		}
-
 		// Validate the auth params and create a token client if we are using
 		// auth
 		if natsJWT != "" || natsNKeySeed != "" {
@@ -171,13 +138,19 @@ can be set using an environment variable named "NATS_NAME_PREFIX"
 			}
 		}
 
+		// Calculate the SHA-1 hash of the config to use as the queue name. This
+		// means that sources with the same config will be in the same queue.
+		// Note that the config object implements redaction in the String()
+		// method so we don't have to worry about leaking secrets
+		configHash := fmt.Sprintf("%x", sha1.Sum([]byte(rc.String())))
+
 		e, err := discovery.NewEngine()
 		if err != nil {
 			log.WithFields(log.Fields{
 				"error": err.Error(),
 			}).Fatal("Error initializing Engine")
 		}
-		e.Name = "source-template"
+		e.Name = "k8s-source"
 		e.NATSOptions = &connect.NATSOptions{
 			NumRetries:        -1,
 			RetryDelay:        5 * time.Second,
@@ -189,11 +162,9 @@ can be set using an environment variable named "NATS_NAME_PREFIX"
 			ReconnectJitter:   2 * time.Second,
 			TokenClient:       tokenClient,
 		}
-		e.NATSQueueName = "source-template" // This should be the same as your engine name
+		e.NATSQueueName = fmt.Sprintf("k8s-source-%v", configHash)
 		e.MaxParallelExecutions = maxParallel
 
-		e.AddSources(sourceList...)
-
 		// Start HTTP server for status
 		healthCheckPort := 8080
 		healthCheckPath := "/healthz"
@@ -223,37 +194,114 @@ can be set using an environment variable named "NATS_NAME_PREFIX"
 			os.Exit(1)
 		}
 
-		err = e.Start()
+		// Create channels for interrupts
+		quit := make(chan os.Signal, 1)
+		signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
+		restart := make(chan watch.Event, 1024)
+
+		// Get the initial starting point
+		list, err := clientSet.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
 
 		if err != nil {
-			log.WithFields(log.Fields{
-				"error": err,
-			}).Error("Could not start engine")
+			log.Fatalf("Could not list namespaces: %v", err)
+		}
 
-			os.Exit(1)
+		// Watch namespaces from here
+		sendInitialEvents := false
+		wi, err := clientSet.CoreV1().Namespaces().Watch(context.Background(), metav1.ListOptions{
+			SendInitialEvents: &sendInitialEvents,
+			ResourceVersion:   list.ResourceVersion,
+		})
+
+		if err != nil {
+			log.Fatalf("Could not start watching namespaces: %v", err)
 		}
 
-		sigs := make(chan os.Signal, 1)
+		watchCtx, watchCancel := context.WithCancel(context.Background())
+		defer watchCancel()
 
-		signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+		go func() {
+			for {
+				select {
+				case event := <-wi.ResultChan():
+					// Restart the engine
+					restart <- event
+				case <-watchCtx.Done():
+					return
+				}
+			}
+		}()
 
-		<-sigs
+		for {
+			// Query all namespaces
+			log.Info("Listing namespaces")
+			list, err := clientSet.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
 
-		log.Info("Stopping engine")
+			if err != nil {
+				log.Fatal(err)
+			}
 
-		err = e.Stop()
+			namespaces := make([]string, len(list.Items))
 
-		if err != nil {
-			log.WithFields(log.Fields{
-				"error": err,
-			}).Error("Could not stop engine")
+			for i := range list.Items {
+				namespaces[i] = list.Items[i].Name
+			}
 
-			os.Exit(1)
-		}
+			log.Infof("got %v namespaces", len(namespaces))
+
+			// Create the sources
+			sourceList := sources.LoadAllSources(clientSet, k8sURL.Host, namespaces)
+
+			// Add sources to the engine
+			e.AddSources(sourceList...)
+
+			// Start the engine
+			err = e.Start()
+
+			if err != nil {
+				log.WithFields(log.Fields{
+					"error": err,
+				}).Error("Could not start engine")
+
+				os.Exit(1)
+			}
+
+			// Start waiting for either an interrupt or a restart
+			select {
+			case <-quit:
+				log.Info("Stopping engine")
+
+				err = e.Stop()
+
+				if err != nil {
+					log.WithFields(log.Fields{
+						"error": err,
+					}).Error("Could not stop engine")
+
+					os.Exit(1)
+				}
 
-		log.Info("Stopped")
+				log.Info("Stopped")
 
-		os.Exit(0)
+				os.Exit(0)
+			case event := <-restart:
+				log.Infof("Restarting engine due to namespace event: %v", event.Type)
+
+				// Stop the engine
+				err = e.Stop()
+
+				if err != nil {
+					log.WithFields(log.Fields{
+						"error": err,
+					}).Error("Could not stop engine")
+
+					os.Exit(1)
+				}
+
+				// Clear the sources
+				e.ClearSources()
+			}
+		}
 	},
 }
 

config.json

@@ -0,0 +1,6 @@
+{
+  "auths": {
+    "ghcr.io": {},
+  },
+  "credsStore": "desktop"
+}

go.mod

@@ -4,11 +4,12 @@ go 1.20
 
 // Direct dependencies of my codebase
 require (
+	github.com/google/uuid v1.3.0
 	github.com/nats-io/jwt/v2 v2.4.1
 	github.com/nats-io/nkeys v0.4.4
-	github.com/overmindtech/connect v0.9.0
-	github.com/overmindtech/discovery v0.19.0
-	github.com/overmindtech/sdp-go v0.29.1
+	github.com/overmindtech/connect v0.10.0
+	github.com/overmindtech/discovery v0.20.2
+	github.com/overmindtech/sdp-go v0.30.0
 	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
@@ -42,7 +43,6 @@ require (
 	github.com/google/gnostic v0.6.9 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/imdario/mergo v0.3.15 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
@@ -58,7 +58,7 @@ require (
 	github.com/nats-io/nats.go v1.25.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/overmindtech/api-client v0.14.0 // indirect
-	github.com/overmindtech/sdpcache v1.3.0 // indirect
+	github.com/overmindtech/sdpcache v1.4.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.7 // indirect
 	github.com/pkg/errors v0.9.1 // indirect