fix: correctly handle stopping services with reverse dependencies

This bug showed up with extension services: say we have a service
`ext-foo` which depends on service `cri`.

Service `ext-foo` will be started correctly only once `cri` is up.

But we should also stop `ext-foo` before `cri` is stopped, as otherwise
the dependency chain is broken. This PR fixes exactly that: once `cri`
is stopped, anything which depends on it should be stopped. We should
stop as well anything which depends on `ext-foo` (transitive
dependency).

In practical terms we use dependency on `cri` in extension service to
correctly stop/start extension services with `/var` filesystem
mount/unmount.

Signed-off-by: Andrey Smirnov <andrey.smirnov@talos-systems.com>
(cherry picked from commit 992e230)

Author: smira

internal/app/machined/pkg/runtime/v1alpha1/v1alpha1_sequencer_tasks.go

@@ -714,7 +714,8 @@ func StartAllServices(seq runtime.Sequence, data interface{}) (runtime.TaskExecu
 // StopServicesForUpgrade represents the StopServicesForUpgrade task.
 func StopServicesForUpgrade(seq runtime.Sequence, data interface{}) (runtime.TaskExecutionFunc, string) {
 	return func(ctx context.Context, logger *log.Logger, r runtime.Runtime) (err error) {
-		return system.Services(nil).StopWithRevDepenencies(ctx, "cri", "etcd", "kubelet", "udevd")
+		// stopping 'cri' service stops everything which depends on it (kubelet, etcd, ...)
+		return system.Services(nil).StopWithRevDepenencies(ctx, "cri", "udevd")
 	}, "stopServicesForUpgrade"
 }
 

internal/app/machined/pkg/system/services/extension.go

@@ -14,7 +14,6 @@ import (
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 
 	"github.com/talos-systems/talos/internal/app/machined/pkg/runtime"
-	"github.com/talos-systems/talos/internal/app/machined/pkg/system"
 	"github.com/talos-systems/talos/internal/app/machined/pkg/system/events"
 	"github.com/talos-systems/talos/internal/app/machined/pkg/system/runner"
 	"github.com/talos-systems/talos/internal/app/machined/pkg/system/runner/containerd"
@@ -66,8 +65,6 @@ func (svc *Extension) Condition(r runtime.Runtime) conditions.Condition {
 
 	for _, dep := range svc.Spec.Depends {
 		switch {
-		case dep.Service != "":
-			conds = append(conds, system.WaitForService(system.StateEventUp, dep.Service))
 		case dep.Path != "":
 			conds = append(conds, conditions.WaitForFileToExist(dep.Path))
 		case len(dep.Network) > 0:
@@ -86,7 +83,15 @@ func (svc *Extension) Condition(r runtime.Runtime) conditions.Condition {
 
 // DependsOn implements the Service interface.
 func (svc *Extension) DependsOn(r runtime.Runtime) []string {
-	return []string{"containerd"}
+	deps := []string{"containerd"}
+
+	for _, dep := range svc.Spec.Depends {
+		if dep.Service != "" {
+			deps = append(deps, dep.Service)
+		}
+	}
+
+	return deps
 }
 
 // Runner implements the Service interface.

internal/app/machined/pkg/system/system.go

@@ -229,7 +229,7 @@ func (s *singleton) Stop(ctx context.Context, serviceIDs ...string) (err error)
 
 // StopWithRevDepenencies will initiate a shutdown of the specified services waiting for reverse dependencies to finish first.
 //
-// If reverse dependency is not stopped, this method might block waiting on it being stopped forever.
+// If reverse dependency is not stopped, this method might block waiting on it being stopped for up to 30 seconds.
 func (s *singleton) StopWithRevDepenencies(ctx context.Context, serviceIDs ...string) (err error) {
 	if len(serviceIDs) == 0 {
 		return
@@ -247,35 +247,77 @@ func (s *singleton) StopWithRevDepenencies(ctx context.Context, serviceIDs ...st
 
 //nolint:gocyclo
 func (s *singleton) stopServices(ctx context.Context, services []string, waitForRevDependencies bool) error {
-	stateCopy := make(map[string]*ServiceRunner)
+	servicesToStop := make(map[string]*ServiceRunner)
 
 	if services == nil {
 		for name, svcrunner := range s.state {
-			stateCopy[name] = svcrunner
+			servicesToStop[name] = svcrunner
 		}
 	} else {
 		for _, name := range services {
 			if _, ok := s.state[name]; !ok {
 				continue
 			}
 
-			stateCopy[name] = s.state[name]
+			servicesToStop[name] = s.state[name]
 		}
 	}
 
-	s.mu.Unlock()
-
-	// build reverse dependencies
+	// build reverse dependencies, and expand the list of services to stop
+	// with services which depend on the one being stopped
 	reverseDependencies := make(map[string][]string)
 
 	if waitForRevDependencies {
-		for name, svcrunner := range stateCopy {
+		// expand the list of services to stop with the list of services which depend
+		// on the ones being stopped
+		// the loop is run as long as more dependencies are added to the list
+		for {
+			expanded := false
+
+			for name, svcrunner := range s.state {
+				if _, scheduledToStop := servicesToStop[name]; scheduledToStop {
+					continue
+				}
+
+				dependencies := svcrunner.service.DependsOn(s.runtime)
+
+				shouldStopService := false
+
+				for _, dependency := range dependencies {
+					for scheduledService := range servicesToStop {
+						if scheduledService == dependency {
+							shouldStopService = true
+
+							break
+						}
+					}
+
+					if shouldStopService {
+						break
+					}
+				}
+
+				if shouldStopService {
+					servicesToStop[name] = svcrunner
+					expanded = true
+				}
+			}
+
+			if !expanded {
+				break
+			}
+		}
+
+		// build a list of dependencies to wait for before stopping each of the services
+		for name, svcrunner := range servicesToStop {
 			for _, dependency := range svcrunner.service.DependsOn(s.runtime) {
 				reverseDependencies[dependency] = append(reverseDependencies[dependency], name)
 			}
 		}
 	}
 
+	s.mu.Unlock()
+
 	// shutdown all the services waiting for rev deps
 	var shutdownWg sync.WaitGroup
 
@@ -285,7 +327,7 @@ func (s *singleton) stopServices(ctx context.Context, services []string, waitFor
 
 	stoppedConds := []conditions.Condition{}
 
-	for name, svcrunner := range stateCopy {
+	for name, svcrunner := range servicesToStop {
 		shutdownWg.Add(1)
 
 		stoppedConds = append(stoppedConds, WaitForService(StateEventDown, name))

internal/app/machined/pkg/system/system_test.go

@@ -27,12 +27,6 @@ func (suite *SystemServicesSuite) TestStartShutdown() {
 	time.Sleep(10 * time.Millisecond)
 
 	suite.Require().NoError(system.Services(nil).Unload(context.Background(), "trustd", "notrunning"))
-
-	system.Services(nil).Shutdown(context.TODO())
-}
-
-func TestSystemServicesSuite(t *testing.T) {
-	suite.Run(t, new(SystemServicesSuite))
 }
 
 func (suite *SystemServicesSuite) TestStartStop() {
@@ -47,3 +41,29 @@ func (suite *SystemServicesSuite) TestStartStop() {
 	)
 	suite.Assert().NoError(err)
 }
+
+func (suite *SystemServicesSuite) TestStopWithRevDeps() {
+	system.Services(nil).LoadAndStart(
+		&MockService{name: "cri"},
+		&MockService{name: "networkd", dependencies: []string{"cri"}},
+		&MockService{name: "vland", dependencies: []string{"networkd"}},
+	)
+	time.Sleep(10 * time.Millisecond)
+
+	// stopping cri should stop all services
+	suite.Require().NoError(system.Services(nil).StopWithRevDepenencies(context.Background(), "cri"))
+
+	// no services should be running
+	for _, name := range []string{"cri", "networkd", "vland"} {
+		svc, running, err := system.Services(nil).IsRunning(name)
+		suite.Require().NoError(err)
+		suite.Assert().NotNil(svc)
+		suite.Assert().False(running)
+	}
+
+	system.Services(nil).Shutdown(context.TODO())
+}
+
+func TestSystemServicesSuite(t *testing.T) {
+	suite.Run(t, new(SystemServicesSuite))
+}