Pipeline KNE cluster init deploys

cmd/deploy/deploy.go

@@ -110,6 +110,7 @@ func deployFn(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	d.Progress = viper.GetBool("progress")
+	d.Timeout = viper.GetDuration("deploy_timeout")
 	d.ReportUsage = viper.GetBool("report_usage")
 	d.ReportUsageProjectID = viper.GetString("report_usage_project_id")
 	d.ReportUsageTopicID = viper.GetString("report_usage_topic_id")

cmd/root.go

@@ -45,6 +45,7 @@ environment.`,
 	root.PersistentFlags().String("report_usage_project_id", "", "Project to report anonymous usage metrics to")
 	root.PersistentFlags().String("report_usage_topic_id", "", "Topic to report anonymous usage metrics to")
 	root.PersistentFlags().Bool("progress", false, "Display progress of container bringup")
+	root.PersistentFlags().Duration("deploy_timeout", 0, "Overall timeout for deploying the ingress, CNI and controllers, 0 for the default")
 	root.PersistentPreRunE = func(cmd *cobra.Command, args []string) error {
 		if *cfgFile == "" {
 			return nil

deploy/deploy.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -28,6 +29,7 @@ import (
 	epb "github.com/openconfig/kne/proto/event"
 	"github.com/pborman/uuid"
 	metallbv1 "go.universe.tf/metallb/api/v1beta1"
+	"golang.org/x/sync/errgroup"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -51,7 +53,23 @@ var (
 	setPIDMaxScript = filepath.Join(homedir.HomeDir(), "kne-internal", "set_pid_max.sh")
 	pullRetryDelay  = time.Second
 	poolRetryDelay  = 5 * time.Second
-	healthTimeout   = time.Minute
+	// healthTimeout is how long a single component is given to become
+	// healthy, measured from the moment that component finished deploying.
+	//
+	// This is deliberately generous.  Operator readiness is mostly a fixed
+	// cost rather than noise, and reproduces closely between runs: MetalLB
+	// has been measured at 69.15s and 69.25s on two separate deployments,
+	// with Lemming at ~45s and IxiaTG at ~44s.  MetalLB therefore never fit
+	// in a one minute budget; it simply was never held to one, because the
+	// wait that matters happens inside its Deploy rather than in Healthy.
+	// defaultDeployTimeout, not this, is the backstop against a deployment
+	// that is truly stuck.
+	healthTimeout = 3 * time.Minute
+	// defaultDeployTimeout bounds the concurrent deployment of the ingress,
+	// CNI and controllers as a whole.  It is a backstop for the cases the
+	// per-component budgets cannot catch, such as a Deploy that hangs: those
+	// run kubectl, which has no timeout of its own.
+	defaultDeployTimeout = 10 * time.Minute
 
 	// Stubs for testing.
 	execLookPath       = exec.LookPath
@@ -88,6 +106,15 @@ type Controller interface {
 	Healthy(context.Context) error
 }
 
+// A component is the part of Ingress, CNI and Controller that a deployment
+// drives: bring yourself up, then report when you are ready.  Ingress, CNI and
+// Controller all satisfy it, which lets them be deployed uniformly and
+// concurrently.
+type component interface {
+	Deploy(context.Context) error
+	Healthy(context.Context) error
+}
+
 type Deployment struct {
 	Cluster     Cluster      `kne:"cluster"`
 	Ingress     Ingress      `kne:"ingress"`
@@ -98,6 +125,12 @@ type Deployment struct {
 	// standard output.
 	Progress bool
 
+	// Timeout bounds the concurrent deployment of the ingress, CNI and
+	// controllers.  It does not cover bringing up the cluster itself, which
+	// happens first and is largely not interruptible.  If zero,
+	// defaultDeployTimeout is used.
+	Timeout time.Duration
+
 	// If ReportUsage is true then anonymous usage metrics will be
 	// published using Cloud PubSub.
 	ReportUsage bool
@@ -255,47 +288,123 @@ func (d *Deployment) Deploy(ctx context.Context, kubecfg string) (rerr error) {
 		}()
 	}
 
+	// Inject the clients into every component up front.  Ingress needs its
+	// client during Deploy, and the others only happen to get away without
+	// one because their Deploy is a plain kubectl apply today.
 	d.Ingress.SetKClient(kClient)
 	d.Ingress.SetRCfg(rCfg)
 	d.Ingress.SetDockerNetworkResourceName(d.Cluster.GetDockerNetworkResourceName())
-
-	log.Infof("Deploying ingress...")
-	if err := d.Ingress.Deploy(ctx); err != nil {
-		return fmt.Errorf("failed to deploy ingress: %w", err)
-	}
-	tCtx, cancel := context.WithTimeout(ctx, healthTimeout)
-	defer cancel()
-	if err := d.Ingress.Healthy(tCtx); err != nil {
-		return fmt.Errorf("failed to check if ingress is healthy: %w", err)
-	}
-	log.Infof("Ingress healthy")
-	log.Infof("Deploying CNI...")
-	if err := d.CNI.Deploy(ctx); err != nil {
-		return fmt.Errorf("failed to deploy CNI: %w", err)
-	}
 	d.CNI.SetKClient(kClient)
-	tCtx, cancel = context.WithTimeout(ctx, healthTimeout)
-	defer cancel()
-	if err := d.CNI.Healthy(tCtx); err != nil {
-		return fmt.Errorf("failed to check if CNI is healthy: %w", err)
-	}
-	log.Infof("CNI healthy")
 	for _, c := range d.Controllers {
-		log.Infof("Deploying controller...")
-		if err := c.Deploy(ctx); err != nil {
-			return fmt.Errorf("failed to deploy controller: %w", err)
-		}
 		c.SetKClient(kClient)
-		tCtx, cancel = context.WithTimeout(ctx, healthTimeout)
-		defer cancel()
-		if err := c.Healthy(tCtx); err != nil {
-			return fmt.Errorf("failed to check if controller is healthy: %w", err)
-		}
 	}
-	log.Infof("Controllers deployed and healthy")
+
+	if err := d.deployComponents(ctx); err != nil {
+		return err
+	}
+	log.Infof("Ingress, CNI and controllers deployed and healthy")
 	return nil
 }
 
+// deployComponents brings up the ingress, CNI and controllers.
+//
+// They are independent of each other: each applies its own manifests and then
+// waits for its own workloads in its own namespace.  Deploying them
+// concurrently makes a deployment cost the slowest component rather than the
+// sum of all of them.
+func (d *Deployment) deployComponents(ctx context.Context) error {
+	dCtx, dCancel := context.WithTimeout(ctx, d.timeout())
+	defer dCancel()
+	g, gCtx := errgroup.WithContext(dCtx)
+
+	start := func(name string, c component) {
+		// A component that was never configured has nothing to deploy.  The
+		// sequential code often never reached a nil component because an
+		// earlier one failed first; running them together always reaches it,
+		// so say so rather than panicking.
+		if c == nil {
+			log.Warningf("No %s configured, skipping", name)
+			return
+		}
+		g.Go(func() error {
+			// Tag any command this component runs, so its output can be
+			// picked out of the interleaved log.
+			cCtx := run.WithLabel(gCtx, name)
+			log.Infof("Deploying %s...", name)
+			if err := c.Deploy(cCtx); err != nil {
+				return componentErr(name, dCtx, gCtx, nil, fmt.Errorf("failed to deploy %s: %w", name, err))
+			}
+			log.Infof("%s deployed", name)
+
+			// Each component gets its own health budget, started when that
+			// component finished deploying.  A single shared deadline would
+			// charge a component for a slow sibling, and would expire for
+			// everything still pending at once, hiding which component was
+			// actually stuck.
+			hCtx, hCancel := context.WithTimeout(gCtx, healthTimeout)
+			defer hCancel()
+			if err := c.Healthy(hCtx); err != nil {
+				return componentErr(name, dCtx, gCtx, hCtx, fmt.Errorf("failed to check if %s is healthy: %w", name, err))
+			}
+			log.Infof("%s healthy", name)
+			return nil
+		})
+	}
+
+	start("ingress", d.Ingress)
+	start("CNI", d.CNI)
+	for _, c := range d.Controllers {
+		start(controllerName(c), c)
+	}
+	return g.Wait()
+}
+
+// timeout returns the budget for the deployment as a whole.
+func (d *Deployment) timeout() time.Duration {
+	if d.Timeout > 0 {
+		return d.Timeout
+	}
+	return defaultDeployTimeout
+}
+
+// controllerName returns a name for c suitable for logs and errors, derived
+// from its type: a *CEOSLabSpec becomes "CEOSLab controller".  Controllers are
+// otherwise anonymous, and with several of them coming up at once "controller"
+// alone does not say which one failed.
+func controllerName(c Controller) string {
+	n := fmt.Sprintf("%T", c)
+	n = n[strings.LastIndex(n, ".")+1:]
+	if n = strings.TrimSuffix(n, "Spec"); n == "" {
+		return "controller"
+	}
+	return n + " controller"
+}
+
+// componentErr explains why a component stopped early.  The components share a
+// context, so the first failure cancels the rest; without this every other
+// component would report an indistinguishable "context canceled" and bury the
+// one real error.
+//
+// overall is the deployment-wide context, group the errgroup context, and own
+// the component's own health context, which is nil while it is still
+// deploying.
+func componentErr(name string, overall, group, own context.Context, err error) error {
+	switch {
+	case errors.Is(overall.Err(), context.DeadlineExceeded):
+		// The whole deployment ran out of time, so this component's own
+		// budget is not the interesting fact.
+		return fmt.Errorf("%s did not finish before the overall deployment timeout expired: %w", name, err)
+	case own != nil && errors.Is(own.Err(), context.DeadlineExceeded):
+		return fmt.Errorf("%s was not healthy within %v: %w", name, healthTimeout, err)
+	case group.Err() != nil:
+		// Either a sibling failed or the deployment was canceled, e.g. by the
+		// pod watcher seeing a container fail to start.
+		return fmt.Errorf("%s abandoned before completing: %w", name, err)
+	default:
+		return err
+	}
+}
+
 func validateKubectlVersion() error {
 	output, err := run.OutCommand("kubectl", "version", "--output=yaml")
 	if err != nil {
@@ -418,27 +527,35 @@ func (d *Deployment) Healthy(ctx context.Context) error {
 		return fmt.Errorf("failed to check cluster is healthy: %w", err)
 	}
 	log.Infof("Cluster healthy")
-	tCtx, cancel := context.WithTimeout(ctx, healthTimeout)
-	defer cancel()
-	if err := d.Ingress.Healthy(tCtx); err != nil {
-		return fmt.Errorf("failed to check ingress is healthy: %w", err)
-	}
-	log.Infof("Ingress healthy")
-	tCtx, cancel = context.WithTimeout(ctx, healthTimeout)
-	defer cancel()
-	if err := d.CNI.Healthy(tCtx); err != nil {
-		return fmt.Errorf("failed to check CNI is healthy: %w", err)
-	}
-	log.Infof("CNI healthy")
+
+	// As in Deploy, the components are independent, so check them
+	// concurrently and give each its own budget.
+	hCtx, hCancel := context.WithTimeout(ctx, d.timeout())
+	defer hCancel()
+	g, gCtx := errgroup.WithContext(hCtx)
+
+	check := func(name string, c component) {
+		if c == nil {
+			log.Warningf("No %s configured, skipping health check", name)
+			return
+		}
+		g.Go(func() error {
+			cCtx, cancel := context.WithTimeout(gCtx, healthTimeout)
+			defer cancel()
+			if err := c.Healthy(cCtx); err != nil {
+				return componentErr(name, hCtx, gCtx, cCtx, fmt.Errorf("failed to check %s is healthy: %w", name, err))
+			}
+			log.Infof("%s healthy", name)
+			return nil
+		})
+	}
+
+	check("ingress", d.Ingress)
+	check("CNI", d.CNI)
 	for _, c := range d.Controllers {
-		tCtx, cancel = context.WithTimeout(ctx, healthTimeout)
-		defer cancel()
-		if err := c.Healthy(tCtx); err != nil {
-			return fmt.Errorf("failed to check controller is healthy: %w", err)
-		}
+		check(controllerName(c), c)
 	}
-	log.Infof("Controllers healthy")
-	return nil
+	return g.Wait()
 }
 
 func init() {
@@ -987,7 +1104,7 @@ func (m *MetalLBSpec) Deploy(ctx context.Context) error {
 		m.Manifest = filepath.Join(m.ManifestDir, "metallb-native.yaml")
 	}
 	log.Infof("Deploying MetalLB from: %s", m.Manifest)
-	if err := run.LogCommand("kubectl", "apply", "-f", m.Manifest); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", m.Manifest); err != nil {
 		return fmt.Errorf("failed to deploy metallb: %w", err)
 	}
 	if _, err := m.kClient.CoreV1().Secrets("metallb-system").Get(ctx, "memberlist", metav1.GetOptions{}); err != nil {
@@ -1120,7 +1237,7 @@ func (m *MeshnetSpec) Deploy(ctx context.Context) error {
 		m.Manifest = filepath.Join(m.ManifestDir, "manifest.yaml")
 	}
 	log.Infof("Deploying Meshnet from: %s", m.Manifest)
-	if err := run.LogCommand("kubectl", "apply", "-f", m.Manifest); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", m.Manifest); err != nil {
 		return fmt.Errorf("failed to deploy meshnet: %w", err)
 	}
 	log.Infof("Meshnet Deployed")
@@ -1197,7 +1314,7 @@ func (c *CEOSLabSpec) Deploy(ctx context.Context) error {
 		c.Operator = filepath.Join(c.ManifestDir, "manifest.yaml")
 	}
 	log.Infof("Deploying CEOSLab controller from: %s", c.Operator)
-	if err := run.LogCommand("kubectl", "apply", "-f", c.Operator); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", c.Operator); err != nil {
 		return fmt.Errorf("failed to deploy ceoslab operator: %w", err)
 	}
 	log.Infof("CEOSLab controller deployed")
@@ -1246,7 +1363,7 @@ func (l *LemmingSpec) Deploy(ctx context.Context) error {
 		l.Operator = filepath.Join(l.ManifestDir, "manifest.yaml")
 	}
 	log.Infof("Deploying Lemming controller from: %s", l.Operator)
-	if err := run.LogCommand("kubectl", "apply", "-f", l.Operator); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", l.Operator); err != nil {
 		return fmt.Errorf("failed to deploy lemming operator: %w", err)
 	}
 	log.Infof("Lemming controller deployed")
@@ -1295,7 +1412,7 @@ func (s *SRLinuxSpec) Deploy(ctx context.Context) error {
 		s.Operator = filepath.Join(s.ManifestDir, "manifest.yaml")
 	}
 	log.Infof("Deploying SRLinux controller from: %s", s.Operator)
-	if err := run.LogCommand("kubectl", "apply", "-f", s.Operator); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", s.Operator); err != nil {
 		return fmt.Errorf("failed to deploy srlinux operator: %w", err)
 	}
 	log.Infof("SRLinux controller deployed")
@@ -1346,7 +1463,7 @@ func (i *IxiaTGSpec) Deploy(ctx context.Context) error {
 		i.Operator = filepath.Join(i.ManifestDir, "ixiatg-operator.yaml")
 	}
 	log.Infof("Deploying IxiaTG controller from: %s", i.Operator)
-	if err := run.LogCommand("kubectl", "apply", "-f", i.Operator); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", i.Operator); err != nil {
 		return fmt.Errorf("failed to deploy ixiatg operator: %w", err)
 	}
 
@@ -1380,7 +1497,7 @@ func (i *IxiaTGSpec) Deploy(ctx context.Context) error {
 		i.ConfigMap = f.Name()
 	}
 	log.Infof("Deploying IxiaTG config map from: %s", i.ConfigMap)
-	if err := run.LogCommand("kubectl", "apply", "-f", i.ConfigMap); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", i.ConfigMap); err != nil {
 		return fmt.Errorf("failed to deploy ixiatg config map: %w", err)
 	}
 	log.Infof("IxiaTG controller deployed")
@@ -1424,7 +1541,7 @@ func (c *CdnosSpec) Deploy(ctx context.Context) error {
 		c.Operator = f.Name()
 	}
 	log.Infof("Deploying Cdnos controller from: %s", c.Operator)
-	if err := run.LogCommand("kubectl", "apply", "-f", c.Operator); err != nil {
+	if err := run.LogCommandContext(ctx, "kubectl", "apply", "-f", c.Operator); err != nil {
 		return fmt.Errorf("failed to deploy cdnos operator: %w", err)
 	}
 	log.Infof("Cdnos controller deployed")