(7.0) Pull only required packages during join (#1862)

lib/expand/builder.go

@@ -39,6 +39,8 @@ type planBuilder struct {
 	Application app.Application
 	// Runtime is the Runtime of the app being installed
 	Runtime app.Application
+	// GravityPackage is the gravity package to install
+	GravityPackage loc.Locator
 	// TeleportPackage is the teleport package to install
 	TeleportPackage loc.Locator
 	// PlanetPackage is the planet package to install
@@ -144,6 +146,13 @@ func (b *planBuilder) AddPullPhase(plan *storage.OperationPlan) {
 			ExecServer:  &b.JoiningNode,
 			Package:     &b.Application.Package,
 			ServiceUser: &b.ServiceUser,
+			Pull: &storage.PullData{
+				Packages: []loc.Locator{
+					b.GravityPackage,
+					b.TeleportPackage,
+					b.PlanetPackage,
+				},
+			},
 		},
 		Requires: []string{installphases.ConfigurePhase, installphases.BootstrapPhase},
 	})
@@ -340,6 +349,11 @@ func (p *Peer) getPlanBuilder(ctx operationContext) (*planBuilder, error) {
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	gravityPackage, err := application.Manifest.Dependencies.ByName(
+		constants.GravityPackage)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
 	teleportPackage, err := application.Manifest.Dependencies.ByName(
 		constants.TeleportPackage)
 	if err != nil {
@@ -375,6 +389,7 @@ func (p *Peer) getPlanBuilder(ctx operationContext) (*planBuilder, error) {
 	return &planBuilder{
 		Application:     *application,
 		Runtime:         *runtime,
+		GravityPackage:  *gravityPackage,
 		TeleportPackage: *teleportPackage,
 		PlanetPackage:   *planetPackage,
 		JoiningNode:     operation.Servers[0],

lib/expand/phases/checks.go

@@ -25,6 +25,8 @@ import (
 	"github.com/gravitational/gravity/lib/ops"
 	"github.com/gravitational/gravity/lib/rpc"
 
+	"github.com/fatih/color"
+	"github.com/gravitational/satellite/agent/proto/agentpb"
 	"github.com/gravitational/trace"
 	"github.com/sirupsen/logrus"
 )
@@ -88,8 +90,22 @@ func (p *checksExecutor) Execute(ctx context.Context) error {
 	// For multi-node checks, use one of master nodes as an "anchor" so
 	// the joining node will be compared against that master (e.g. for
 	// the OS check, time drift check, etc).
-	failed := checker.CheckNode(ctx, *node)
-	failed = append(failed, checker.CheckNodes(ctx, []checks.Server{*master, *node})...)
+	probes := checker.CheckNode(ctx, *node)
+	probes = append(probes, checker.CheckNodes(ctx, []checks.Server{*master, *node})...)
+	// Sort probes out into warnings and real failures.
+	var failed []*agentpb.Probe
+	for _, probe := range probes {
+		if probe.Status != agentpb.Probe_Failed {
+			continue
+		}
+		if probe.Severity == agentpb.Probe_Warning {
+			p.Progress.NextStep(color.YellowString(probe.Detail))
+		}
+		if probe.Severity == agentpb.Probe_Critical {
+			p.Progress.NextStep(color.RedString(probe.Detail))
+			failed = append(failed, probe)
+		}
+	}
 	if len(failed) != 0 {
 		return trace.BadParameter("The following checks failed:\n%v",
 			checks.FormatFailedChecks(failed))

lib/expand/plan_test.go

@@ -51,6 +51,7 @@ type PlanSuite struct {
 	regularAgent    *storage.LoginEntry
 	teleportPackage *loc.Locator
 	planetPackage   *loc.Locator
+	gravityPackage  *loc.Locator
 	appPackage      loc.Locator
 	serviceUser     storage.OSUser
 	cluster         *ops.Site
@@ -77,6 +78,8 @@ func (s *PlanSuite) SetUpSuite(c *check.C) {
 	c.Assert(err, check.IsNil)
 	s.teleportPackage, err = app.Manifest.Dependencies.ByName(constants.TeleportPackage)
 	c.Assert(err, check.IsNil)
+	s.gravityPackage, err = app.Manifest.Dependencies.ByName(constants.GravityPackage)
+	c.Assert(err, check.IsNil)
 	s.dnsConfig = storage.DNSConfig{
 		Addrs: []string{"127.0.0.3"},
 		Port:  10053,
@@ -282,6 +285,13 @@ func (s *PlanSuite) verifyPullPhase(c *check.C, phase storage.OperationPhase) {
 			ExecServer:  &s.joiningNode,
 			Package:     &s.appPackage,
 			ServiceUser: &s.serviceUser,
+			Pull: &storage.PullData{
+				Packages: []loc.Locator{
+					*s.gravityPackage,
+					*s.teleportPackage,
+					*s.planetPackage,
+				},
+			},
 		},
 		Requires: []string{installphases.ConfigurePhase, installphases.BootstrapPhase},
 	}, phase)

lib/install/phases/pull.go

@@ -30,6 +30,7 @@ import (
 	"github.com/gravitational/gravity/lib/pack"
 	"github.com/gravitational/gravity/lib/schema"
 	"github.com/gravitational/gravity/lib/state"
+	"github.com/gravitational/gravity/lib/storage"
 	"github.com/gravitational/gravity/lib/systeminfo"
 	"github.com/gravitational/gravity/lib/utils"
 
@@ -45,6 +46,9 @@ func NewPull(p fsm.ExecutorParams, operator ops.Operator, wizardPack, localPack
 	if p.Phase.Data == nil || p.Phase.Data.ServiceUser == nil {
 		return nil, trace.BadParameter("service user is required")
 	}
+	if p.Phase.Data.Pull == nil {
+		return nil, trace.BadParameter("phase does not contain pull data")
+	}
 
 	serviceUser, err := systeminfo.UserFromOSUser(*p.Phase.Data.ServiceUser)
 	if err != nil {
@@ -79,6 +83,7 @@ func NewPull(p fsm.ExecutorParams, operator ops.Operator, wizardPack, localPack
 		LocalApps:      localApps,
 		ExecutorParams: p,
 		ServiceUser:    *serviceUser,
+		Pull:           *p.Phase.Data.Pull,
 		runtimePackage: *runtimePackage,
 		remote:         remote,
 	}, nil
@@ -97,6 +102,8 @@ type pullExecutor struct {
 	LocalApps app.Applications
 	// ServiceUser is the user used for services and system storage
 	ServiceUser systeminfo.User
+	// Pull contains applications and packages to pull
+	Pull storage.PullData
 	// ExecutorParams is common executor params
 	fsm.ExecutorParams
 	// remote specifies the server remote control interface
@@ -107,11 +114,19 @@ type pullExecutor struct {
 
 // Execute executes the pull phase
 func (p *pullExecutor) Execute(ctx context.Context) error {
-	err := p.pullUserApplication()
-	if err != nil {
-		return trace.Wrap(err)
+	if len(p.Pull.Packages) != 0 {
+		err := p.pullPackages(p.Pull.Packages)
+		if err != nil {
+			return trace.Wrap(err)
+		}
 	}
-	err = p.pullConfiguredPackages()
+	if len(p.Pull.Apps) != 0 {
+		err := p.pullApps(p.Pull.Apps)
+		if err != nil {
+			return trace.Wrap(err)
+		}
+	}
+	err := p.pullConfiguredPackages()
 	if err != nil {
 		return trace.Wrap(err)
 	}
@@ -137,22 +152,40 @@ func (p *pullExecutor) Execute(ctx context.Context) error {
 	return nil
 }
 
-func (p *pullExecutor) pullUserApplication() error {
-	p.Progress.NextStep("Pulling user application")
-	p.Info("Pulling user application.")
-	// TODO do not pull user app on regular nodes
-	// FIXME: use context to promptly abort the pull
-	_, err := service.PullApp(service.AppPullRequest{
-		FieldLogger: p.FieldLogger,
-		SrcPack:     p.WizardPackages,
-		DstPack:     p.LocalPackages,
-		SrcApp:      p.WizardApps,
-		DstApp:      p.LocalApps,
-		Package:     *p.Phase.Data.Package,
-	})
-	// Ignore already exists as the steps need to be re-entrant
-	if err != nil && !trace.IsAlreadyExists(err) {
-		return trace.Wrap(err)
+func (p *pullExecutor) pullPackages(locators []loc.Locator) error {
+	p.Progress.NextStep("Pulling packages")
+	p.Infof("Pulling packages: %v.", locators)
+	for _, locator := range locators {
+		p.Progress.NextStep("Pulling package %v:%v", locator.Name, locator.Version)
+		_, err := service.PullPackage(service.PackagePullRequest{
+			FieldLogger: p.FieldLogger,
+			SrcPack:     p.WizardPackages,
+			DstPack:     p.LocalPackages,
+			Package:     locator,
+		})
+		if err != nil && !trace.IsAlreadyExists(err) { // Must be re-entrant.
+			return trace.Wrap(err)
+		}
+	}
+	return nil
+}
+
+func (p *pullExecutor) pullApps(locators []loc.Locator) error {
+	p.Progress.NextStep("Pulling applications")
+	p.Infof("Pulling applications: %v.", locators)
+	for _, locator := range locators {
+		p.Progress.NextStep("Pulling application %v:%v", locator.Name, locator.Version)
+		_, err := service.PullApp(service.AppPullRequest{
+			FieldLogger: p.FieldLogger,
+			SrcPack:     p.WizardPackages,
+			DstPack:     p.LocalPackages,
+			SrcApp:      p.WizardApps,
+			DstApp:      p.LocalApps,
+			Package:     locator,
+		})
+		if err != nil && !trace.IsAlreadyExists(err) { // Must be re-entrant.
+			return trace.Wrap(err)
+		}
 	}
 	return nil
 }

lib/install/planbuilder.go

@@ -55,6 +55,8 @@ type PlanBuilder struct {
 	RBACPackage loc.Locator
 	// GravitySitePackage is the gravity-site app package
 	GravitySitePackage loc.Locator
+	// GravityPackage is the gravity binary package
+	GravityPackage loc.Locator
 	// DNSAppPackage is the dns-app app package
 	DNSAppPackage loc.Locator
 	// Masters is the list of master nodes
@@ -205,7 +207,7 @@ func (b *PlanBuilder) AddBootstrapPhase(plan *storage.OperationPlan) {
 }
 
 // AddPullPhase appends package download phase to the provided plan
-func (b *PlanBuilder) AddPullPhase(plan *storage.OperationPlan) {
+func (b *PlanBuilder) AddPullPhase(plan *storage.OperationPlan) error {
 	var pullPhases []storage.OperationPhase
 	allNodes := append(b.Masters, b.Nodes...)
 	for i, node := range allNodes {
@@ -215,6 +217,10 @@ func (b *PlanBuilder) AddPullPhase(plan *storage.OperationPlan) {
 		} else {
 			description = "Pull packages on node %v"
 		}
+		pullData, err := b.getPullData(node)
+		if err != nil {
+			return trace.Wrap(err)
+		}
 		pullPhases = append(pullPhases, storage.OperationPhase{
 			ID:          fmt.Sprintf("%v/%v", phases.PullPhase, node.Hostname),
 			Description: fmt.Sprintf(description, node.Hostname),
@@ -223,6 +229,7 @@ func (b *PlanBuilder) AddPullPhase(plan *storage.OperationPlan) {
 				ExecServer:  &allNodes[i],
 				Package:     &b.Application.Package,
 				ServiceUser: &b.ServiceUser,
+				Pull:        pullData,
 			},
 			Requires: fsm.RequireIfPresent(plan, phases.ConfigurePhase, phases.BootstrapPhase),
 			Step:     3,
@@ -236,6 +243,34 @@ func (b *PlanBuilder) AddPullPhase(plan *storage.OperationPlan) {
 		Parallel:    true,
 		Step:        3,
 	})
+	return nil
+}
+
+// getPullData returns package and application locators that should be pulled
+// during the operation on the provided node.
+func (b *PlanBuilder) getPullData(node storage.Server) (*storage.PullData, error) {
+	// Master nodes pull the entire application to be able to invoke an
+	// install hook from any master node local state.
+	if node.ClusterRole == string(schema.ServiceRoleMaster) {
+		return &storage.PullData{
+			Apps: []loc.Locator{
+				b.Application.Package,
+			},
+		}, nil
+	}
+	// Regular nodes pull only packages required for runtime such as planet
+	// or teleport. The planet package also depends on the node role.
+	planetPackage, err := b.Application.Manifest.RuntimePackageForProfile(node.Role)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return &storage.PullData{
+		Packages: []loc.Locator{
+			b.GravityPackage,
+			b.TeleportPackage,
+			*planetPackage,
+		},
+	}, nil
 }
 
 // AddMastersPhase appends master nodes system installation phase to the provided plan
@@ -667,6 +702,11 @@ func (c *Config) GetPlanBuilder(operator ops.Operator, cluster ops.Site, op ops.
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
+	gravityPackage, err := cluster.App.Manifest.Dependencies.ByName(
+		constants.GravityPackage)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
 	dnsAppPackage, err := cluster.App.Manifest.Dependencies.ByName(
 		constants.DNSAppPackage)
 	if err != nil {
@@ -704,6 +744,7 @@ func (c *Config) GetPlanBuilder(operator ops.Operator, cluster ops.Site, op ops.
 		TeleportPackage:    *teleportPackage,
 		RBACPackage:        *rbacPackage,
 		GravitySitePackage: *gravitySitePackage,
+		GravityPackage:     *gravityPackage,
 		DNSAppPackage:      *dnsAppPackage,
 		Masters:            masters,
 		Nodes:              nodes,

lib/install/reconfigure/plan_test.go

@@ -259,6 +259,11 @@ func (s *ReconfiguratorSuite) verifyPullPhase(c *check.C, phase storage.Operatio
 					ExecServer:  &master,
 					Package:     s.suite.Package(),
 					ServiceUser: &s.suite.Cluster().ServiceUser,
+					Pull: &storage.PullData{
+						Apps: []loc.Locator{
+							*(s.suite.Package()),
+						},
+					},
 				},
 				Requires: []string{installphases.ConfigurePhase},
 			},

lib/install/suite.go

@@ -58,6 +58,7 @@ type PlanSuite struct {
 	adminAgent         *storage.LoginEntry
 	regularAgent       *storage.LoginEntry
 	teleportPackage    *loc.Locator
+	gravityPackage     *loc.Locator
 	runtimePackage     loc.Locator
 	rbacPackage        *loc.Locator
 	runtimeApplication *loc.Locator
@@ -117,6 +118,8 @@ func (s *PlanSuite) SetUpSuite(c *check.C) {
 	c.Assert(err, check.IsNil)
 	s.teleportPackage, err = app.Manifest.Dependencies.ByName(constants.TeleportPackage)
 	c.Assert(err, check.IsNil)
+	s.gravityPackage, err = app.Manifest.Dependencies.ByName(constants.GravityPackage)
+	c.Assert(err, check.IsNil)
 	runtimePackage, err := app.Manifest.DefaultRuntimePackage()
 	c.Assert(err, check.IsNil)
 	s.runtimePackage = *runtimePackage
@@ -362,6 +365,11 @@ func (s *PlanSuite) VerifyPullPhase(c *check.C, phase storage.OperationPhase) {
 					ExecServer:  &s.masterNode,
 					Package:     &s.installer.config.App.Package,
 					ServiceUser: serviceUser,
+					Pull: &storage.PullData{
+						Apps: []loc.Locator{
+							s.installer.config.App.Package,
+						},
+					},
 				},
 				Requires: []string{phases.ConfigurePhase, phases.BootstrapPhase},
 			},
@@ -372,6 +380,13 @@ func (s *PlanSuite) VerifyPullPhase(c *check.C, phase storage.OperationPhase) {
 					ExecServer:  &s.regularNode,
 					Package:     &s.installer.config.App.Package,
 					ServiceUser: serviceUser,
+					Pull: &storage.PullData{
+						Packages: []loc.Locator{
+							*s.gravityPackage,
+							*s.teleportPackage,
+							s.runtimePackage,
+						},
+					},
 				},
 				Requires: []string{phases.ConfigurePhase, phases.BootstrapPhase},
 			},

lib/storage/plan.go

@@ -155,6 +155,8 @@ type OperationPhaseData struct {
 	ServiceUser *OSUser `json:"service_user,omitempty" yaml:"service_user,omitempty"`
 	// Data is arbitrary text data to provide to a phase executor
 	Data string `json:"data,omitempty" yaml:"data,omitempty"`
+	// Pull contains applications and packages that should be pulled
+	Pull *PullData `json:"pull,omitempty" yaml:"pull,omitempty"`
 	// GarbageCollect specifies configuration specific to garbage collect operation
 	GarbageCollect *GarbageCollectOperationData `json:"garbage_collect,omitempty" yaml:"garbage_collect,omitempty"`
 	// Update specifies configuration specific to update operations
@@ -163,6 +165,14 @@ type OperationPhaseData struct {
 	Install *InstallOperationData `json:"install,omitempty" yaml:"install,omitempty"`
 }
 
+// PullData contains applications and packages to pull
+type PullData struct {
+	// Packages is a list of packages to pull
+	Packages []loc.Locator `json:"packages,omitempty" yaml:"packages,omitempty"`
+	// Apps is a list of applications to pull
+	Apps []loc.Locator `json:"apps,omitempty" yaml:"apps,omitempty"`
+}
+
 // ElectionChange describes changes to make to cluster elections
 type ElectionChange struct {
 	// EnableServers is a list of servers that we should enable elections on