Plumb in top-level context

Plumbs the top-level context into the agent, create, & gather commands
and into the asset graph.

Author: patrickdillon

cmd/openshift-install/agent.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"context"
+
 	"github.com/spf13/cobra"
 
 	"github.com/openshift/installer/cmd/openshift-install/agent"
@@ -14,7 +16,7 @@ import (
 	"github.com/openshift/installer/pkg/asset/password"
 )
 
-func newAgentCmd() *cobra.Command {
+func newAgentCmd(ctx context.Context) *cobra.Command {
 	agentCmd := &cobra.Command{
 		Use:   "agent",
 		Short: "Commands for supporting cluster installation using agent installer",
@@ -23,7 +25,7 @@ func newAgentCmd() *cobra.Command {
 		},
 	}
 
-	agentCmd.AddCommand(newAgentCreateCmd())
+	agentCmd.AddCommand(newAgentCreateCmd(ctx))
 	agentCmd.AddCommand(agent.NewWaitForCmd())
 	agentCmd.AddCommand(newAgentGraphCmd())
 	return agentCmd
@@ -115,8 +117,7 @@ var (
 	agentTargets = []target{agentConfigTarget, agentManifestsTarget, agentImageTarget, agentPXEFilesTarget, agentConfigImageTarget, agentUnconfiguredIgnitionTarget}
 )
 
-func newAgentCreateCmd() *cobra.Command {
-
+func newAgentCreateCmd(ctx context.Context) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "create",
 		Short: "Commands for generating agent installation artifacts",
@@ -127,7 +128,7 @@ func newAgentCreateCmd() *cobra.Command {
 
 	for _, t := range agentTargets {
 		t.command.Args = cobra.ExactArgs(0)
-		t.command.Run = runTargetCmd(t.assets...)
+		t.command.Run = runTargetCmd(ctx, t.assets...)
 		cmd.AddCommand(t.command)
 	}
 

cmd/openshift-install/create.go

@@ -128,12 +128,10 @@ var (
 			Short: "Create an OpenShift cluster",
 			// FIXME: add longer descriptions for our commands with examples for better UX.
 			// Long:  "",
-			PostRun: func(_ *cobra.Command, _ []string) {
-				// Setup a context that is canceled when the user presses Ctrl+C,
-				// or SIGTERM and SIGINT are received, this allows for a clean shutdown.
-				ctx, cancel := context.WithCancel(context.TODO())
-				defer cancel()
-				logrus.RegisterExitHandler(cancel)
+			PostRun: func(cmd *cobra.Command, _ []string) {
+
+				// Get the context that was set in newCreateCmd.
+				ctx := cmd.Context()
 
 				exitCode, err := clusterCreatePostRun(ctx)
 				if err != nil {
@@ -168,7 +166,7 @@ func clusterCreatePostRun(ctx context.Context) (int, error) {
 	}
 
 	// Handle the case when the API server is not reachable.
-	if err := handleUnreachableAPIServer(config); err != nil {
+	if err := handleUnreachableAPIServer(ctx, config); err != nil {
 		logrus.Fatal(fmt.Errorf("unable to handle api server override: %w", err))
 	}
 
@@ -177,7 +175,7 @@ func clusterCreatePostRun(ctx context.Context) (int, error) {
 	//
 	timer.StartTimer("Bootstrap Complete")
 	if err := waitForBootstrapComplete(ctx, config); err != nil {
-		bundlePath, gatherErr := runGatherBootstrapCmd(command.RootOpts.Dir)
+		bundlePath, gatherErr := runGatherBootstrapCmd(ctx, command.RootOpts.Dir)
 		if gatherErr != nil {
 			logrus.Error("Attempted to gather debug logs after installation failure: ", gatherErr)
 		}
@@ -278,7 +276,7 @@ func newClientError(errorInfo error) *clusterCreateError {
 	}
 }
 
-func newCreateCmd() *cobra.Command {
+func newCreateCmd(ctx context.Context) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "create",
 		Short: "Create part of an OpenShift cluster",
@@ -289,22 +287,25 @@ func newCreateCmd() *cobra.Command {
 
 	for _, t := range targets {
 		t.command.Args = cobra.ExactArgs(0)
-		t.command.Run = runTargetCmd(t.assets...)
+		t.command.Run = runTargetCmd(ctx, t.assets...)
 		cmd.AddCommand(t.command)
 	}
 
 	return cmd
 }
 
-func runTargetCmd(targets ...asset.WritableAsset) func(cmd *cobra.Command, args []string) {
+func runTargetCmd(ctx context.Context, targets ...asset.WritableAsset) func(cmd *cobra.Command, args []string) {
 	runner := func(directory string) error {
 		fetcher := assetstore.NewAssetsFetcher(directory)
-		return fetcher.FetchAndPersist(targets)
+		return fetcher.FetchAndPersist(ctx, targets)
 	}
 
 	return func(cmd *cobra.Command, args []string) {
 		timer.StartTimer(timer.TotalTimeElapsed)
 
+		// Set the context to be used in the PostRun function.
+		cmd.SetContext(ctx)
+
 		cleanup := command.SetupFileHook(command.RootOpts.Dir)
 		defer cleanup()
 
@@ -856,15 +857,15 @@ func meetsStabilityThreshold(progressing *configv1.ClusterOperatorStatusConditio
 	return progressing.Status == configv1.ConditionFalse && time.Since(progressing.LastTransitionTime.Time).Seconds() > coStabilityThreshold
 }
 
-func handleUnreachableAPIServer(config *rest.Config) error {
+func handleUnreachableAPIServer(ctx context.Context, config *rest.Config) error {
 	assetStore, err := assetstore.NewStore(command.RootOpts.Dir)
 	if err != nil {
 		return fmt.Errorf("failed to create asset store: %w", err)
 	}
 
 	// Ensure that the install is expecting the user to provision their own DNS solution.
 	installConfig := &installconfig.InstallConfig{}
-	if err := assetStore.Fetch(installConfig); err != nil {
+	if err := assetStore.Fetch(ctx, installConfig); err != nil {
 		return fmt.Errorf("failed to fetch %s: %w", installConfig.Name(), err)
 	}
 	switch installConfig.Config.Platform.Name() { //nolint:gocritic
@@ -877,7 +878,7 @@ func handleUnreachableAPIServer(config *rest.Config) error {
 	}
 
 	lbConfig := &lbconfig.Config{}
-	if err := assetStore.Fetch(lbConfig); err != nil {
+	if err := assetStore.Fetch(ctx, lbConfig); err != nil {
 		return fmt.Errorf("failed to fetch %s: %w", lbConfig.Name(), err)
 	}
 

cmd/openshift-install/gather.go

@@ -34,7 +34,7 @@ import (
 	_ "github.com/openshift/installer/pkg/gather/gcp"
 )
 
-func newGatherCmd() *cobra.Command {
+func newGatherCmd(ctx context.Context) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "gather",
 		Short: "Gather debugging data for a given installation failure",
@@ -47,7 +47,7 @@ to debug the installation failures`,
 			return cmd.Help()
 		},
 	}
-	cmd.AddCommand(newGatherBootstrapCmd())
+	cmd.AddCommand(newGatherBootstrapCmd(ctx))
 	return cmd
 }
 
@@ -58,15 +58,15 @@ var gatherBootstrapOpts struct {
 	skipAnalysis bool
 }
 
-func newGatherBootstrapCmd() *cobra.Command {
+func newGatherBootstrapCmd(ctx context.Context) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "bootstrap",
 		Short: "Gather debugging data for a failing-to-bootstrap control plane",
 		Args:  cobra.ExactArgs(0),
 		Run: func(_ *cobra.Command, _ []string) {
 			cleanup := command.SetupFileHook(command.RootOpts.Dir)
 			defer cleanup()
-			bundlePath, err := runGatherBootstrapCmd(command.RootOpts.Dir)
+			bundlePath, err := runGatherBootstrapCmd(ctx, command.RootOpts.Dir)
 			if err != nil {
 				logrus.Fatal(err)
 			}
@@ -87,14 +87,14 @@ func newGatherBootstrapCmd() *cobra.Command {
 	return cmd
 }
 
-func runGatherBootstrapCmd(directory string) (string, error) {
+func runGatherBootstrapCmd(ctx context.Context, directory string) (string, error) {
 	assetStore, err := assetstore.NewStore(directory)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to create asset store")
 	}
 	// add the default bootstrap key pair to the sshKeys list
 	bootstrapSSHKeyPair := &tls.BootstrapSSHKeyPair{}
-	if err := assetStore.Fetch(bootstrapSSHKeyPair); err != nil {
+	if err := assetStore.Fetch(ctx, bootstrapSSHKeyPair); err != nil {
 		return "", errors.Wrapf(err, "failed to fetch %s", bootstrapSSHKeyPair.Name())
 	}
 	tmpfile, err := os.CreateTemp("", "bootstrap-ssh")
@@ -118,7 +118,7 @@ func runGatherBootstrapCmd(directory string) (string, error) {
 
 	if ha.Bootstrap == "" && len(ha.Masters) == 0 {
 		config := &installconfig.InstallConfig{}
-		if err := assetStore.Fetch(config); err != nil {
+		if err := assetStore.Fetch(ctx, config); err != nil {
 			return "", errors.Wrapf(err, "failed to fetch %s", config.Name())
 		}
 

cmd/openshift-install/main.go

@@ -44,17 +44,17 @@ func installerMain() {
 	logrus.RegisterExitHandler(shutdown)
 
 	for _, subCmd := range []*cobra.Command{
-		newCreateCmd(),
+		newCreateCmd(ctx),
 		newDestroyCmd(),
 		newWaitForCmd(),
-		newGatherCmd(),
+		newGatherCmd(ctx),
 		newAnalyzeCmd(),
 		newVersionCmd(),
 		newGraphCmd(),
 		newCoreOSCmd(),
 		newCompletionCmd(),
 		newExplainCmd(),
-		newAgentCmd(),
+		newAgentCmd(ctx),
 	} {
 		rootCmd.AddCommand(subCmd)
 	}

pkg/asset/cluster/cluster.go

@@ -42,6 +42,7 @@ type Cluster struct {
 }
 
 var _ asset.WritableAsset = (*Cluster)(nil)
+var _ asset.Generator = (*Cluster)(nil)
 
 // Name returns the human-friendly name of the asset.
 func (c *Cluster) Name() string {
@@ -78,7 +79,7 @@ func (c *Cluster) Dependencies() []asset.Asset {
 }
 
 // Generate launches the cluster and generates the terraform state file on disk.
-func (c *Cluster) Generate(parents asset.Parents) (err error) {
+func (c *Cluster) GenerateWithContext(ctx context.Context, parents asset.Parents) (err error) {
 	if InstallDir == "" {
 		logrus.Fatalf("InstallDir has not been set for the %q asset", c.Name())
 	}
@@ -136,7 +137,7 @@ func (c *Cluster) Generate(parents asset.Parents) (err error) {
 	if err != nil {
 		return fmt.Errorf("error getting infrastructure provider: %w", err)
 	}
-	files, err := provider.Provision(InstallDir, parents)
+	files, err := provider.Provision(ctx, InstallDir, parents)
 	if files != nil {
 		c.FileList = append(c.FileList, files...) // append state files even in case of failure
 	}
@@ -165,3 +166,9 @@ func (c *Cluster) Load(f asset.FileFetcher) (found bool, err error) {
 
 	return false, nil
 }
+
+// Generate is implemented so the Cluster Asset maintains compatibility
+// with the Asset interface. It should never be called.
+func (c *Cluster) Generate(_ asset.Parents) (err error) {
+	panic("Cluster.Generate was called instead of Cluster.GenerateWithContext")
+}

pkg/asset/store.go

@@ -1,11 +1,15 @@
 package asset
 
+import (
+	"context"
+)
+
 // Store is a store for the states of assets.
 type Store interface {
 	// Fetch retrieves the state of the given asset, generating it and its
 	// dependencies if necessary. When purging consumed assets, none of the
 	// assets in assetsToPreserve will be purged.
-	Fetch(assetToFetch Asset, assetsToPreserve ...WritableAsset) error
+	Fetch(ctx context.Context, assetToFetch Asset, assetsToPreserve ...WritableAsset) error
 
 	// Destroy removes the asset from all its internal state and also from
 	// disk if possible.

pkg/asset/store/assetsfetcher.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -12,7 +13,7 @@ import (
 // AssetsFetcher it's used to retrieve and resolve a specified set of assets.
 type AssetsFetcher interface {
 	// Fetchs and persists all the writable assets from the configured assets store.
-	FetchAndPersist(assets []asset.WritableAsset) error
+	FetchAndPersist(context.Context, []asset.WritableAsset) error
 }
 
 type fetcher struct {
@@ -36,14 +37,14 @@ func asFileWriter(a asset.WritableAsset) asset.FileWriter {
 }
 
 // Fetchs all the writable assets from the configured assets store.
-func (f *fetcher) FetchAndPersist(assets []asset.WritableAsset) error {
+func (f *fetcher) FetchAndPersist(ctx context.Context, assets []asset.WritableAsset) error {
 	assetStore, err := NewStore(f.storeDir)
 	if err != nil {
 		return fmt.Errorf("failed to create asset store: %w", err)
 	}
 
 	for _, a := range assets {
-		err := assetStore.Fetch(a, assets...)
+		err := assetStore.Fetch(ctx, a, assets...)
 		if err != nil {
 			err = errors.Wrapf(err, "failed to fetch %s", a.Name())
 		}

pkg/asset/store/store.go

@@ -1,6 +1,7 @@
 package store
 
 import (
+	"context"
 	"encoding/json"
 	"os"
 	"path/filepath"
@@ -72,8 +73,8 @@ func newStore(dir string) (*storeImpl, error) {
 // Fetch retrieves the state of the given asset, generating it and its
 // dependencies if necessary. When purging consumed assets, none of the
 // assets in preserved will be purged.
-func (s *storeImpl) Fetch(a asset.Asset, preserved ...asset.WritableAsset) error {
-	if err := s.fetch(a, ""); err != nil {
+func (s *storeImpl) Fetch(ctx context.Context, a asset.Asset, preserved ...asset.WritableAsset) error {
+	if err := s.fetch(ctx, a, ""); err != nil {
 		return err
 	}
 	if err := s.saveStateFile(); err != nil {
@@ -192,7 +193,7 @@ func (s *storeImpl) saveStateFile() error {
 // fetch populates the given asset, generating it and its dependencies if
 // necessary, and returns whether or not the asset had to be regenerated and
 // any errors.
-func (s *storeImpl) fetch(a asset.Asset, indent string) error {
+func (s *storeImpl) fetch(ctx context.Context, a asset.Asset, indent string) error {
 	logrus.Debugf("%sFetching %s...", indent, a.Name())
 
 	assetState, ok := s.assets[reflect.TypeOf(a)]
@@ -217,13 +218,13 @@ func (s *storeImpl) fetch(a asset.Asset, indent string) error {
 	dependencies := a.Dependencies()
 	parents := make(asset.Parents, len(dependencies))
 	for _, d := range dependencies {
-		if err := s.fetch(d, increaseIndent(indent)); err != nil {
+		if err := s.fetch(ctx, d, increaseIndent(indent)); err != nil {
 			return errors.Wrapf(err, "failed to fetch dependency of %q", a.Name())
 		}
 		parents.Add(d)
 	}
 	logrus.Debugf("%sGenerating %s...", indent, a.Name())
-	if err := a.Generate(parents); err != nil {
+	if err := asAssetGenerator(a).GenerateWithContext(ctx, parents); err != nil {
 		return errors.Wrapf(err, "failed to generate asset %q", a.Name())
 	}
 	assetState.asset = a
@@ -371,3 +372,14 @@ func (s *storeImpl) Load(a asset.Asset) (asset.Asset, error) {
 
 	return s.assets[reflect.TypeOf(a)].asset, nil
 }
+
+// asAssetGenerator determines if an asset implements the
+// Generate with context function, or if it needs an adapter.
+func asAssetGenerator(a asset.Asset) asset.Generator {
+	switch v := a.(type) {
+	case asset.Generator:
+		return v
+	default:
+		return asset.NewDefaultGenerator(a)
+	}
+}

pkg/infrastructure/aws/sdk/aws.go

@@ -61,7 +61,7 @@ type output struct {
 }
 
 // Provision creates cluster infrastructure using AWS SDK calls.
-func (a InfraProvider) Provision(dir string, parents asset.Parents) ([]*asset.File, error) {
+func (a InfraProvider) Provision(ctx context.Context, dir string, parents asset.Parents) ([]*asset.File, error) {
 	terraformVariables := &tfvarsAsset.TerraformVariables{}
 	parents.Get(terraformVariables)
 	// Unmarshall input from tf variables, so we can use it along with
@@ -109,7 +109,7 @@ func (a InfraProvider) Provision(dir string, parents asset.Parents) ([]*asset.Fi
 	usePublicEndpoints := clusterAWSConfig.PublishStrategy == "External"
 
 	logger := logrus.StandardLogger()
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Minute)
 	defer cancel()
 
 	ec2Client := ec2.New(awsSession)