feat: add kubernetes connection in exec connections (#1181)

* feat: add kubernetes connection in exec connections

* feat: kubernetes Connection object

* fromConfigItem in exec connection

* feat: use the config's scraper

* handle connection namespace

* feat: use the scrape config directly to support not CRD based scrape
configs

* chore: address review comments

connection/environment.go

@@ -1,6 +1,7 @@
 package connection
 
 import (
+	"encoding/json"
 	"fmt"
 	"math/rand"
 	"os"
@@ -9,15 +10,22 @@ import (
 
 	"github.com/flanksource/commons/logger"
 	"github.com/flanksource/duty/context"
+	"github.com/flanksource/duty/models"
+	"github.com/samber/lo"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	textTemplate "text/template"
 )
 
 // +kubebuilder:object:generate=true
 type ExecConnections struct {
-	AWS   *AWSConnection   `yaml:"aws,omitempty" json:"aws,omitempty"`
-	GCP   *GCPConnection   `yaml:"gcp,omitempty" json:"gcp,omitempty"`
-	Azure *AzureConnection `yaml:"azure,omitempty" json:"azure,omitempty"`
+	FromConfigItem *string `yaml:"fromConfigItem,omitempty" json:"fromConfigItem,omitempty" template:"true"`
+
+	Kubernetes *KubernetesConnection `yaml:"kubernetes,omitempty" json:"kubernetes,omitempty"`
+	AWS        *AWSConnection        `yaml:"aws,omitempty" json:"aws,omitempty"`
+	GCP        *GCPConnection        `yaml:"gcp,omitempty" json:"gcp,omitempty"`
+	Azure      *AzureConnection      `yaml:"azure,omitempty" json:"azure,omitempty"`
 }
 
 func saveConfig(configTemplate *textTemplate.Template, view any) (string, error) {
@@ -43,8 +51,9 @@ func saveConfig(configTemplate *textTemplate.Template, view any) (string, error)
 }
 
 var (
-	awsConfigTemplate    *textTemplate.Template
-	gcloudConfigTemplate *textTemplate.Template
+	awsConfigTemplate        *textTemplate.Template
+	kubernetesConfigTemplate *textTemplate.Template
+	gcloudConfigTemplate     *textTemplate.Template
 )
 
 func init() {
@@ -55,20 +64,92 @@ aws_secret_access_key = {{.SecretKey.ValueStatic}}
 `))
 
 	gcloudConfigTemplate = textTemplate.Must(textTemplate.New("").Parse(`{{.Credentials}}`))
+
+	kubernetesConfigTemplate = textTemplate.Must(textTemplate.New("").Parse(`{{.KubeConfig.ValueStatic}}`))
 }
 
-// SetupCConnections creates the necessary credential files and injects env vars
+// SetupConnections creates the necessary credential files and injects env vars
 // into the cmd
-func SetupConnection(ctx context.Context, connections ExecConnections, cmd *osExec.Cmd) error {
+func SetupConnection(ctx context.Context, connections ExecConnections, cmd *osExec.Cmd) (func() error, error) {
+	var cleaner = func() error {
+		return nil
+	}
+
+	if lo.FromPtr(connections.FromConfigItem) != "" {
+		var scraperNamespace string
+		var scraperSpec map[string]any
+
+		{
+			var configItem models.ConfigItem
+			if err := ctx.DB().Where("id = ?", *connections.FromConfigItem).First(&configItem).Error; err != nil {
+				return nil, fmt.Errorf("failed to get config (%s): %w", *connections.FromConfigItem, err)
+			}
+
+			var scrapeConfig models.ConfigScraper
+			if err := ctx.DB().Where("id = ?", lo.FromPtr(configItem.ScraperID)).First(&scrapeConfig).Error; err != nil {
+				return nil, fmt.Errorf("failed to get scrapeconfig (%s): %w", lo.FromPtr(configItem.ScraperID), err)
+			}
+			scraperNamespace = scrapeConfig.Namespace
+
+			if err := json.Unmarshal([]byte(scrapeConfig.Spec), &scraperSpec); err != nil {
+				return nil, fmt.Errorf("unable to unmarshal scrapeconfig spec (id=%s)", *configItem.ScraperID)
+			}
+		}
+
+		if kubernetesScrapers, found, err := unstructured.NestedSlice(scraperSpec, "spec", "kubernetes"); err != nil {
+			return nil, err
+		} else if found {
+			for _, kscraper := range kubernetesScrapers {
+				if kubeconfig, found, err := unstructured.NestedMap(kscraper.(map[string]any), "kubeconfig"); err != nil {
+					return nil, err
+				} else if found {
+					connections.Kubernetes = &KubernetesConnection{}
+					if err := runtime.DefaultUnstructuredConverter.FromUnstructured(kubeconfig, &connections.Kubernetes.KubeConfig); err != nil {
+						return nil, err
+					}
+
+					if err := connections.Kubernetes.Populate(ctx.WithNamespace(scraperNamespace)); err != nil {
+						return nil, fmt.Errorf("failed to hydrate kubernetes connection: %w", err)
+					}
+
+					break
+				}
+			}
+		}
+	}
+
+	if connections.Kubernetes != nil {
+		if lo.FromPtr(connections.FromConfigItem) == "" {
+			if err := connections.Kubernetes.Populate(ctx); err != nil {
+				return nil, fmt.Errorf("failed to hydrate kubernetes connection: %w", err)
+			}
+		}
+
+		configPath, err := saveConfig(kubernetesConfigTemplate, connections.Kubernetes)
+		if err != nil {
+			return nil, fmt.Errorf("failed to store kubernetes credentials: %w", err)
+		}
+
+		cleaner = func() error {
+			return os.RemoveAll(filepath.Dir(configPath))
+		}
+
+		cmd.Env = os.Environ()
+		cmd.Env = append(cmd.Env, fmt.Sprintf("KUBECONFIG=%s", configPath))
+	}
+
 	if connections.AWS != nil {
 		if err := connections.AWS.Populate(ctx); err != nil {
-			return fmt.Errorf("failed to hydrate aws connection: %w", err)
+			return nil, fmt.Errorf("failed to hydrate aws connection: %w", err)
 		}
 
 		configPath, err := saveConfig(awsConfigTemplate, connections.AWS)
-		defer os.RemoveAll(filepath.Dir(configPath))
 		if err != nil {
-			return fmt.Errorf("failed to store AWS credentials: %w", err)
+			return nil, fmt.Errorf("failed to store AWS credentials: %w", err)
+		}
+
+		cleaner = func() error {
+			return os.RemoveAll(filepath.Dir(configPath))
 		}
 
 		cmd.Env = os.Environ()
@@ -81,37 +162,40 @@ func SetupConnection(ctx context.Context, connections ExecConnections, cmd *osEx
 
 	if connections.Azure != nil {
 		if err := connections.Azure.HydrateConnection(ctx); err != nil {
-			return fmt.Errorf("failed to hydrate connection %w", err)
+			return nil, fmt.Errorf("failed to hydrate connection %w", err)
 		}
 
 		// login with service principal
 		runCmd := osExec.Command("az", "login", "--service-principal", "--username", connections.Azure.ClientID.ValueStatic, "--password", connections.Azure.ClientSecret.ValueStatic, "--tenant", connections.Azure.TenantID)
 		if err := runCmd.Run(); err != nil {
-			return fmt.Errorf("failed to login: %w", err)
+			return nil, fmt.Errorf("failed to login: %w", err)
 		}
 	}
 
 	if connections.GCP != nil {
 		if err := connections.GCP.HydrateConnection(ctx); err != nil {
-			return fmt.Errorf("failed to hydrate connection %w", err)
+			return nil, fmt.Errorf("failed to hydrate connection %w", err)
 		}
 
 		configPath, err := saveConfig(gcloudConfigTemplate, connections.GCP)
-		defer os.RemoveAll(filepath.Dir(configPath))
 		if err != nil {
-			return fmt.Errorf("failed to store gcloud credentials: %w", err)
+			return nil, fmt.Errorf("failed to store gcloud credentials: %w", err)
+		}
+
+		cleaner = func() error {
+			return os.RemoveAll(filepath.Dir(configPath))
 		}
 
 		// to configure gcloud CLI to use the service account specified in GOOGLE_APPLICATION_CREDENTIALS,
 		// we need to explicitly activate it
 		runCmd := osExec.Command("gcloud", "auth", "activate-service-account", "--key-file", configPath)
 		if err := runCmd.Run(); err != nil {
-			return fmt.Errorf("failed to activate GCP service account: %w", err)
+			return nil, fmt.Errorf("failed to activate GCP service account: %w", err)
 		}
 
 		cmd.Env = os.Environ()
 		cmd.Env = append(cmd.Env, fmt.Sprintf("GOOGLE_APPLICATION_CREDENTIALS=%s", configPath))
 	}
 
-	return nil
+	return cleaner, nil
 }

connection/kubernetes.go

@@ -0,0 +1,42 @@
+package connection
+
+import (
+	"fmt"
+
+	"github.com/flanksource/duty/models"
+	"github.com/flanksource/duty/types"
+)
+
+// +kubebuilder:object:generate=true
+type KubernetesConnection struct {
+	ConnectionName string       `json:"connection,omitempty"`
+	KubeConfig     types.EnvVar `json:"kubeconfig,omitempty"`
+}
+
+func (t KubernetesConnection) ToModel() models.Connection {
+	return models.Connection{
+		Type:        models.ConnectionTypeKubernetes,
+		Certificate: t.KubeConfig.ValueStatic,
+	}
+}
+
+func (t *KubernetesConnection) Populate(ctx ConnectionContext) error {
+	if t.ConnectionName != "" {
+		connection, err := ctx.HydrateConnectionByURL(t.ConnectionName)
+		if err != nil {
+			return err
+		} else if connection == nil {
+			return fmt.Errorf("connection[%s] not found", t.ConnectionName)
+		}
+
+		t.KubeConfig.ValueStatic = connection.Certificate
+	}
+
+	if v, err := ctx.GetEnvValueFromCache(t.KubeConfig, ctx.GetNamespace()); err != nil {
+		return err
+	} else {
+		t.KubeConfig.ValueStatic = v
+	}
+
+	return nil
+}

shell/shell.go

@@ -107,8 +107,14 @@ func Run(ctx context.Context, exec Exec) (*ExecDetails, error) {
 		cmd.Dir = envParams.mountPoint
 	}
 
-	if err := connection.SetupConnection(ctx, exec.Connections, cmd); err != nil {
+	if cleanup, err := connection.SetupConnection(ctx, exec.Connections, cmd); err != nil {
 		return nil, ctx.Oops().Wrap(err)
+	} else {
+		defer func() {
+			if err := cleanup(); err != nil {
+				logger.Errorf("failed to cleanup connection artifacts: %v", err)
+			}
+		}()
 	}
 
 	envParams.cmd = cmd