Merge pull request #75 from KannanThiru/develop-wait-for-resources

add logic for waiting  for dependent resources

Author: darkstarmv

README.md

@@ -18,6 +18,36 @@ Manages Helm charts running in Kubernetes clusters.
 * Share chart overrides across clusters with a `default.yaml` file.
 * Make cluster-specific chart overrides when necessary.
 
+### Managing Dependencies on Addons sequence (Optional feature)
+
+* Next consecutive installation fails due previous pods take time to come up
+* Implemented Wait for Resource feature before moving to the next pipeline
+    - WaitforDeployment
+    - WaitforDaemonset
+    - WaitforStatefulset
+* May need to collect desired resource getting installed using `helm template` when dependency as needed for continuous execution of pipeline.
+* `Kubectlfiles` options enabled in case some external configuration needed for components outside of helm install
+
+```yaml
+name: cluster1-lab
+releases:
+#
+  - name: sample-server
+    namespace: kube-system
+    version: 3.9.0
+    chartPath: stable/sample-server
+    waitforDeployment: 
+      - sample-server
+      - sample-server-2
+    waitforDaemonset: 
+      - sample-server-ds
+    waitforStatefulset:
+      - sample-db-server
+    kubectlFiles:
+      - sample-server.yaml
+      - sample-server-lab
+```
+
 ### Other features
 * Use it as a [Drone](https://drone.io/) plugin for CI/CD.
 * Read secrets from environment variables.

plugin.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"time"
 
 	"github.com/target/impeller/constants"
 	"github.com/target/impeller/types"
@@ -19,6 +20,14 @@ import (
 
 const (
 	kubectlBin = "kubectl"
+
+	maxRetriesDeployment   = 30
+	maxRetriesDaemonSet    = 30
+	maxRetriesStatefulSet  = 120
+	retryDelayDeployment   = 10 * time.Second
+	retryDelayDaemonSet    = 10 * time.Second
+	retryDelayStatefulSet  = 10 * time.Second
+	statefulSetLogInterval = 3
 )
 
 var (
@@ -154,14 +163,31 @@ func (p *Plugin) updateHelmRepos() error {
 
 func (p *Plugin) installAddon(release *types.Release) error {
 	log.Println("Installing addon:", release.Name, "@", release.Version)
+	var err error
 	switch release.DeploymentMethod {
 	case "kubectl":
-		return p.installAddonViaKubectl(release)
+		err = p.installAddonViaKubectl(release)
 	case "helm":
 		fallthrough
 	default:
-		return p.installAddonViaHelm(release)
+		err = p.installAddonViaHelm(release)
+	}
+	
+	if err != nil {
+		return err
 	}
+	
+	// Wait for resources to be ready
+	if err := p.waitForResources(release); err != nil {
+		return err
+	}
+	
+	// Apply additional kubectl files after resources are ready
+	if err := p.applyKubectlFiles(release); err != nil {
+		return err
+	}
+	
+	return nil
 }
 
 // installAddonViaHelm installs addons via helm upgrade --install RELEASE CHART
@@ -181,6 +207,11 @@ func (p *Plugin) installAddonViaHelm(release *types.Release) error {
 		} else if p.ClusterConfig.Helm.DefaultHistory > 0 {
 			cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "history-max", Value: fmt.Sprint(p.ClusterConfig.Helm.DefaultHistory)})
 		}
+		// Force recreate resources if immutable fields change
+		if release.Force {
+			log.Println("Force flag enabled: will recreate resources with immutable field changes")
+			cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: "--force"})
+		}
 	}
 	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: release.Name})
 	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: release.ChartPath})
@@ -194,6 +225,10 @@ func (p *Plugin) installAddonViaHelm(release *types.Release) error {
 	// Add namespaces to command
 	if release.Namespace != "" {
 		cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "namespace", Value: release.Namespace})
+		// Create namespace if it doesn't exist (won't fail if it already exists like kube-system)
+		if !p.Diffrun {
+			cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: "--create-namespace"})
+		}
 	}
 
 	if p.ClusterConfig.Helm.LogLevel != 0 {
@@ -271,6 +306,243 @@ func (p *Plugin) installAddonViaKubectl(release *types.Release) error {
 	return nil
 }
 
+// waitForResources waits for deployments, daemonsets, and statefulsets to be ready
+func (p *Plugin) waitForResources(release *types.Release) error {
+	// Skip waiting if dry-run or diff-run
+	if p.Dryrun || p.Diffrun {
+		return nil
+	}
+
+	// Wait for Deployments
+	for _, deployment := range release.WaitforDeployment {
+		log.Printf("Waiting for Deployment: %s", deployment)
+		if err := p.waitForResource("deployment", deployment, release.Namespace); err != nil {
+			return fmt.Errorf("error waiting for deployment \"%s\": %v", deployment, err)
+		}
+	}
+
+	// Wait for DaemonSets
+	for _, daemonset := range release.WaitforDaemonSet {
+		log.Printf("Waiting for DaemonSet: %s", daemonset)
+		if err := p.waitForResource("daemonset", daemonset, release.Namespace); err != nil {
+			return fmt.Errorf("error waiting for daemonset \"%s\": %v", daemonset, err)
+		}
+	}
+
+	// Wait for StatefulSets
+	for _, statefulset := range release.WaitforStatefulSet {
+		log.Printf("Waiting for StatefulSet: %s", statefulset)
+		if err := p.waitForResource("statefulset", statefulset, release.Namespace); err != nil {
+			return fmt.Errorf("error waiting for statefulset \"%s\": %v", statefulset, err)
+		}
+	}
+
+	return nil
+}
+
+// waitForResource checks if a resource is ready using polling (read-only operation)
+func (p *Plugin) waitForResource(resourceType, resourceName, namespace string) error {
+	switch resourceType {
+	case "deployment":
+		return p.waitForDeployment(resourceName, namespace)
+	case "daemonset":
+		return p.waitForDaemonSet(resourceName, namespace)
+	case "statefulset":
+		return p.waitForStatefulSet(resourceName, namespace)
+	default:
+		return fmt.Errorf("unsupported resource type: %s", resourceType)
+	}
+}
+
+func (p *Plugin) waitForDeployment(resourceName, namespace string) error {
+	log.Printf("⏳ Waiting for deployment %s/%s to be ready...", namespace, resourceName)
+
+	for i := 0; i < maxRetriesDeployment; i++ {
+		output, err := p.kubectlGetJSONPath("deployment", resourceName, namespace, "{.status.conditions[?(@.type=='Available')].status}")
+		if err == nil && strings.TrimSpace(output) == "True" {
+			log.Printf("✅ Deployment %s/%s is ready", namespace, resourceName)
+			return nil
+		}
+
+		if i < maxRetriesDeployment-1 {
+			time.Sleep(retryDelayDeployment)
+		}
+	}
+
+	return fmt.Errorf("timeout waiting for deployment %s/%s", namespace, resourceName)
+}
+
+func (p *Plugin) waitForDaemonSet(resourceName, namespace string) error {
+	log.Printf("⏳ Waiting for daemonset %s/%s to be ready...", namespace, resourceName)
+
+	for i := 0; i < maxRetriesDaemonSet; i++ {
+		output, err := p.kubectlGetJSONPath("daemonset", resourceName, namespace, "{.status.numberReady},{.status.desiredNumberScheduled}")
+		if err == nil {
+			parts := strings.Split(strings.TrimSpace(output), ",")
+			if len(parts) == 2 && parts[0] == parts[1] && parts[0] != "0" {
+				log.Printf("✅ DaemonSet %s/%s is ready", namespace, resourceName)
+				return nil
+			}
+		}
+
+		if i < maxRetriesDaemonSet-1 {
+			time.Sleep(retryDelayDaemonSet)
+		}
+	}
+
+	return fmt.Errorf("timeout waiting for daemonset %s/%s", namespace, resourceName)
+}
+
+func (p *Plugin) waitForStatefulSet(resourceName, namespace string) error {
+	log.Printf("⏳ Waiting for statefulset %s/%s to be ready...", namespace, resourceName)
+	log.Printf("  (This may take up to %d minutes for larger clusters)", maxRetriesStatefulSet*int(retryDelayStatefulSet.Seconds())/60)
+
+	for i := 0; i < maxRetriesStatefulSet; i++ {
+		output, err := p.kubectlGetJSONPath("statefulset", resourceName, namespace, "{.status.readyReplicas},{.status.replicas}")
+		if err == nil {
+			parts := strings.Split(strings.TrimSpace(output), ",")
+			if len(parts) == 2 && parts[0] == parts[1] && parts[0] != "0" {
+				log.Printf("✅ StatefulSet %s/%s is ready (%s/%s replicas)", namespace, resourceName, parts[0], parts[1])
+				return nil
+			}
+			if len(parts) == 2 && i%statefulSetLogInterval == 0 {
+				log.Printf("  Progress: %s/%s replicas ready (attempt %d/%d)", parts[0], parts[1], i+1, maxRetriesStatefulSet)
+			}
+		}
+
+		if i < maxRetriesStatefulSet-1 {
+			time.Sleep(retryDelayStatefulSet)
+		}
+	}
+
+	return fmt.Errorf("timeout waiting for statefulset %s/%s after %d minutes", namespace, resourceName, maxRetriesStatefulSet*int(retryDelayStatefulSet.Seconds())/60)
+}
+
+func (p *Plugin) kubectlGetJSONPath(resourceType, resourceName, namespace, jsonPath string) (string, error) {
+	cb := commandbuilder.CommandBuilder{Name: constants.KubectlBin}
+	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: "get"})
+	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: resourceType})
+	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: resourceName})
+
+	if namespace != "" {
+		cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "namespace", Value: namespace})
+	}
+
+	if p.KubeContext != "" {
+		cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "context", Value: p.KubeContext})
+	}
+
+	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "output", Value: fmt.Sprintf("jsonpath=%s", jsonPath)})
+
+	output, err := cb.Command().Output()
+	if err != nil {
+		return "", err
+	}
+
+	return string(output), nil
+}
+
+// applyKubectlFiles applies additional kubectl manifest files after deployment
+// Supports both individual files and directories (applies all .yaml/.yml files in directory)
+func (p *Plugin) applyKubectlFiles(release *types.Release) error {
+	// Skip if dry-run or diff-run
+	if p.Dryrun || p.Diffrun {
+		return nil
+	}
+
+	// Skip if no kubectl files are specified
+	if len(release.KubectlFiles) == 0 {
+		return nil
+	}
+
+	log.Println("Applying additional kubectl files for:", release.Name)
+
+	for _, path := range release.KubectlFiles {
+		// Check if path is a file or directory
+		fileInfo, err := os.Stat(path)
+		if err != nil {
+			return fmt.Errorf("error accessing path \"%s\": %v", path, err)
+		}
+
+		var filesToApply []string
+		if fileInfo.IsDir() {
+			// If it's a directory, get all YAML files in it
+			log.Printf("Processing directory: %s", path)
+			files, err := p.getYAMLFilesFromDir(path)
+			if err != nil {
+				return fmt.Errorf("error reading directory \"%s\": %v", path, err)
+			}
+			filesToApply = files
+		} else {
+			// If it's a file, apply it directly
+			filesToApply = []string{path}
+		}
+
+		// Apply each file
+		for _, file := range filesToApply {
+			log.Printf("Applying kubectl file: %s", file)
+			
+			cb := commandbuilder.CommandBuilder{Name: constants.KubectlBin}
+			cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: "apply"})
+			cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "filename", Value: file})
+			
+			// Don't force namespace - let the manifest define its own namespace
+			// This allows resources to be created in their specified namespaces
+			
+			if p.KubeContext != "" {
+				cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeLongParam, Name: "context", Value: p.KubeContext})
+			}
+
+			if err := cb.Run(); err != nil {
+				return fmt.Errorf("error applying kubectl file \"%s\": %v", file, err)
+			}
+			
+			log.Printf("Successfully applied kubectl file: %s", file)
+		}
+	}
+
+	return nil
+}
+
+// getYAMLFilesFromDir returns all .yaml and .yml files from a directory
+// Excludes kustomization.yaml and Kustomization.yaml files
+func (p *Plugin) getYAMLFilesFromDir(dirPath string) ([]string, error) {
+	var yamlFiles []string
+	
+	files, err := ioutil.ReadDir(dirPath)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, file := range files {
+		if file.IsDir() {
+			continue
+		}
+		
+		fileName := file.Name()
+		
+		// Skip kustomization files
+		if fileName == "kustomization.yaml" || fileName == "Kustomization.yaml" || 
+		   fileName == "kustomization.yml" || fileName == "Kustomization.yml" {
+			log.Printf("Skipping kustomization file: %s", fileName)
+			continue
+		}
+		
+		if strings.HasSuffix(fileName, ".yaml") || strings.HasSuffix(fileName, ".yml") {
+			fullPath := dirPath + "/" + fileName
+			yamlFiles = append(yamlFiles, fullPath)
+		}
+	}
+
+	if len(yamlFiles) == 0 {
+		log.Printf("WARNING: No YAML files found in directory: %s", dirPath)
+	} else {
+		log.Printf("Found %d YAML file(s) in directory: %s", len(yamlFiles), dirPath)
+	}
+
+	return yamlFiles, nil
+}
+
 func (p *Plugin) fetchChart(release *types.Release) error {
 	cb := commandbuilder.CommandBuilder{Name: constants.HelmBin}
 	cb.Add(commandbuilder.Arg{Type: commandbuilder.ArgTypeRaw, Value: "fetch"})