feat(region): pod containers start in dependency order (#23669)

pkg/apis/container.go

@@ -144,6 +144,10 @@ type ContainerSpec struct {
 	StartupProbe  *ContainerProbe `json:"startup_probe,omitempty"`
 	AlwaysRestart bool            `json:"always_restart"`
 	Primary       bool            `json:"primary"`
+
+	// DependsOn is a list of container name which this container depends on when pod start
+	// Only works for containers created & started by pod-create & server-start
+	DependsOn []string `json:"depends_on,omitempty"`
 }
 
 func (c *ContainerSpec) NeedProbe() bool {

pkg/compute/guestdrivers/pod.go

@@ -40,6 +40,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
 	"yunion.io/x/onecloud/pkg/compute/models"
 	"yunion.io/x/onecloud/pkg/compute/options"
+	"yunion.io/x/onecloud/pkg/compute/utils"
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/util/pod/remotecommand/spdy"
@@ -89,6 +90,7 @@ func (p *SPodDriver) ValidateCreateData(ctx context.Context, userCred mcclient.T
 
 	ctrNames := sets.NewString()
 	volUniqNames := sets.NewString()
+
 	for idx, ctr := range input.Pod.Containers {
 		if err := p.validateContainerData(ctx, userCred, idx, input.Name, ctr, input); err != nil {
 			return nil, errors.Wrapf(err, "data of %d container", idx)
@@ -109,6 +111,15 @@ func (p *SPodDriver) ValidateCreateData(ctx context.Context, userCred mcclient.T
 		}
 	}
 
+	err := utils.TopologicalSortContainers(
+		input.Pod.Containers,
+		func(ctr *api.PodContainerCreateInput) string { return ctr.Name },
+		func(ctr *api.PodContainerCreateInput) []string { return ctr.DependsOn },
+	)
+	if err != nil {
+		return nil, errors.Wrap(err, "invalid container dependency")
+	}
+
 	return input, nil
 }
 

pkg/compute/models/containers.go

@@ -38,6 +38,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
 	"yunion.io/x/onecloud/pkg/compute/options"
+	"yunion.io/x/onecloud/pkg/compute/utils"
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
@@ -205,6 +206,27 @@ func (m *SContainerManager) ValidateSpec(ctx context.Context, userCred mcclient.
 		return errors.Wrap(err, "validate probe configuration")
 	}
 
+	if ctr != nil {
+		// only detect loop when update container
+		ctrs, err := m.GetContainersByPod(pod.GetId())
+		if err != nil {
+			return errors.Wrap(err, "get containers by pod")
+		}
+		for idx, container := range ctrs {
+			if container.GetId() == ctr.GetId() {
+				ctrs[idx].Spec = spec
+			}
+		}
+		err = utils.TopologicalSortContainers(
+			ctrs,
+			func(ctr SContainer) string { return ctr.Name },
+			func(ctr SContainer) []string { return ctr.Spec.DependsOn },
+		)
+		if err != nil {
+			return errors.Wrap(err, "validate topological sort")
+		}
+	}
+
 	return nil
 }
 

pkg/compute/tasks/guest/pod_create_task.go

@@ -62,7 +62,7 @@ func (t *PodCreateTask) OnPodCreated(ctx context.Context, guest *models.SGuest,
 	}
 
 	for idx, ctr := range ctrs {
-		if err := ctr.StartCreateTask(ctx, t.GetUserCred(), t.GetTaskId(), t.GetParams()); err != nil {
+		if err := ctr.StartCreateTask(ctx, t.GetUserCred(), t.GetTaskId(), nil); err != nil {
 			t.onCreateContainerError(ctx, guest, errors.Wrapf(err, "start container %d creation task", idx))
 			return
 		}
@@ -96,8 +96,18 @@ func (t *PodCreateTask) OnContainerCreated(ctx context.Context, guest *models.SG
 		}
 	}
 	if isAllCreated {
-		t.SetStage("OnStatusSynced", nil)
-		guest.StartSyncstatus(ctx, t.GetUserCred(), t.GetTaskId())
+		if jsonutils.QueryBoolean(t.GetParams(), "auto_start", false) {
+			t.SetStage("OnContainerStarted", nil)
+			task, err := taskman.TaskManager.NewTask(ctx, "PodStartContainerInDependencyOrderTask", guest, t.GetUserCred(), nil, t.GetTaskId(), "", nil)
+			if err != nil {
+				t.SetStageFailed(ctx, jsonutils.NewString(errors.Wrap(err, "New PodStartContainerInDependencyOrderTask").Error()))
+				return
+			}
+			task.ScheduleRun(nil)
+		} else {
+			t.SetStage("OnStatusSynced", nil)
+			guest.StartSyncstatus(ctx, t.GetUserCred(), t.GetTaskId())
+		}
 	}
 }
 
@@ -106,6 +116,16 @@ func (t *PodCreateTask) OnContainerCreatedFailed(ctx context.Context, guest *mod
 	t.SetStageFailed(ctx, data)
 }
 
+func (t *PodCreateTask) OnContainerStartedFailed(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) {
+	guest.SetStatus(ctx, t.GetUserCred(), api.POD_STATUS_CREATE_CONTAINER_FAILED, data.String())
+	t.SetStageFailed(ctx, data)
+}
+
+func (t *PodCreateTask) OnContainerStarted(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) {
+	t.SetStage("OnStatusSynced", nil)
+	guest.StartSyncstatus(ctx, t.GetUserCred(), t.GetTaskId())
+}
+
 func (t *PodCreateTask) OnStatusSynced(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) {
 	t.SetStageComplete(ctx, nil)
 }

pkg/compute/tasks/guest/pod_start_container_in_dependency_order_task.go

@@ -0,0 +1,92 @@
+package guest
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
+	"yunion.io/x/onecloud/pkg/compute/models"
+	"yunion.io/x/onecloud/pkg/compute/utils"
+)
+
+type PodStartContainerInDependencyOrderTask struct {
+	SGuestBaseTask
+}
+
+func init() {
+	taskman.RegisterTask(PodStartContainerInDependencyOrderTask{})
+}
+
+func (task *PodStartContainerInDependencyOrderTask) taskFailed(ctx context.Context, pod *models.SGuest, err string) {
+	task.SetStageFailed(ctx, jsonutils.NewString(err))
+}
+
+func (task *PodStartContainerInDependencyOrderTask) taskComplete(ctx context.Context, pod *models.SGuest) {
+	task.SetStageComplete(ctx, nil)
+}
+
+func (t *PodStartContainerInDependencyOrderTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body jsonutils.JSONObject) {
+	pod := obj.(*models.SGuest)
+	ctrs, err := models.GetContainerManager().GetContainersByPod(pod.GetId())
+	if err != nil {
+		t.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
+		return
+	}
+
+	// init graph
+	dep, err := utils.NewDependencyTopoGraph(
+		ctrs,
+		func(ctr models.SContainer) string { return ctr.Id },
+		func(ctr models.SContainer) string { return ctr.Name },
+		func(ctr models.SContainer) []string { return ctr.Spec.DependsOn },
+	)
+	if err != nil {
+		t.taskFailed(ctx, pod, errors.Wrap(err, "New pod container dependency").Error())
+		return
+	}
+
+	t.SaveParams(jsonutils.Marshal(dep).(*jsonutils.JSONDict))
+
+	t.requestContainersStart(ctx, pod, nil)
+}
+
+func (t *PodStartContainerInDependencyOrderTask) requestContainersStart(ctx context.Context, pod *models.SGuest, body jsonutils.JSONObject) {
+	dep := new(utils.DependencyTopoGraph[models.SContainer])
+	if err := t.GetParams().Unmarshal(dep); err != nil {
+		t.taskFailed(ctx, pod, errors.Wrap(err, "Unmarshal container order").Error())
+		return
+	}
+
+	fetchById := func(uuid string) models.SContainer {
+		pctr, err := models.GetContainerManager().FetchById(uuid)
+		if err != nil {
+			log.Infof("FetchById %s error: %s", uuid, err.Error())
+			return models.SContainer{}
+		}
+		return *pctr.(*models.SContainer)
+	}
+	currentBatch := dep.GetNextBatch(fetchById)
+	if currentBatch == nil {
+		t.taskComplete(ctx, pod)
+		return
+	}
+
+	// start current Batch
+	t.SetStage("OnContainerStarted", jsonutils.Marshal(dep).(*jsonutils.JSONDict))
+	if err := models.GetContainerManager().StartBatchStartTask(ctx, t.GetUserCred(), currentBatch, t.GetId()); err != nil {
+		t.OnContainerStartedFailed(ctx, pod, jsonutils.NewString(err.Error()))
+		return
+	}
+}
+
+func (t *PodStartContainerInDependencyOrderTask) OnContainerStarted(ctx context.Context, pod *models.SGuest, data jsonutils.JSONObject) {
+	t.requestContainersStart(ctx, pod, nil)
+}
+
+func (t *PodStartContainerInDependencyOrderTask) OnContainerStartedFailed(ctx context.Context, pod *models.SGuest, data jsonutils.JSONObject) {
+	t.taskFailed(ctx, pod, data.String())
+}

pkg/compute/tasks/guest/pod_start_task.go

@@ -45,16 +45,19 @@ func (t *PodStartTask) OnInit(ctx context.Context, obj db.IStandaloneModel, body
 
 func (t *PodStartTask) OnPodStarted(ctx context.Context, pod *models.SGuest, _ jsonutils.JSONObject) {
 	pod.SetStatus(ctx, t.GetUserCred(), api.POD_STATUS_STARTING_CONTAINER, "")
-	ctrs, err := models.GetContainerManager().GetContainersByPod(pod.GetId())
-	if err != nil {
-		t.OnContainerStartedFailed(ctx, pod, jsonutils.NewString(errors.Wrap(err, "GetContainersByPod").Error()))
-		return
-	}
+	// ctrs, err := models.GetContainerManager().GetContainersByPod(pod.GetId())
+	// if err != nil {
+	// 	t.OnContainerStartedFailed(ctx, pod, jsonutils.NewString(errors.Wrap(err, "GetContainersByPod").Error()))
+	// 	return
+	// }
 	t.SetStage("OnContainerStarted", nil)
-	if err := models.GetContainerManager().StartBatchStartTask(ctx, t.GetUserCred(), ctrs, t.GetId()); err != nil {
-		t.OnContainerStartedFailed(ctx, pod, jsonutils.NewString(err.Error()))
+
+	task, err := taskman.TaskManager.NewTask(ctx, "PodStartContainerInDependencyOrderTask", pod, t.GetUserCred(), nil, t.GetTaskId(), "", nil)
+	if err != nil {
+		t.SetStageFailed(ctx, jsonutils.NewString(errors.Wrap(err, "New PodStartContainerInDependencyOrderTask").Error()))
 		return
 	}
+	task.ScheduleRun(nil)
 }
 
 func (t *PodStartTask) OnPodStartedFailed(ctx context.Context, pod *models.SGuest, reason jsonutils.JSONObject) {

pkg/compute/utils/doc.go

@@ -0,0 +1 @@
+package utils // import "yunion.io/x/onecloud/pkg/compute/utils"

pkg/compute/utils/pod_container_dependencies.go

@@ -0,0 +1,137 @@
+package utils
+
+import (
+	"yunion.io/x/pkg/errors"
+)
+
+type GetObjIdName[T any] func(T) string
+type GetDependencies[T any] func(T) []string
+
+func TopologicalSortContainers[T any](objs []T, getName GetObjIdName[T], getDependencies GetDependencies[T]) error {
+	if len(objs) == 0 {
+		return nil
+	}
+
+	// Build a dependency graph and an in-degree table
+	graph := make(map[string][]string)
+	inDegree := make(map[string]int)
+
+	// init graph and inDegree
+	for _, obj := range objs {
+		inDegree[getName(obj)] = 0
+	}
+	for _, obj := range objs {
+		oName := getName(obj)
+		for _, dep := range getDependencies(obj) {
+			if _, exists := inDegree[dep]; !exists {
+				return errors.Errorf("The dependent container %s does not exist.", dep)
+			}
+			graph[dep] = append(graph[dep], oName)
+			inDegree[oName]++
+		}
+	}
+
+	// Topological sorting: use a queue to process nodes with an in-degree of 0
+	queue := []string{}
+	for name, degree := range inDegree {
+		if degree == 0 {
+			queue = append(queue, name)
+		}
+	}
+
+	sorted := []string{}
+	for len(queue) > 0 {
+		current := queue[0]
+		queue = queue[1:]
+		sorted = append(sorted, current)
+
+		// Decrease the in-degree of neighboring nodes
+		for _, neighbor := range graph[current] {
+			inDegree[neighbor]--
+			if inDegree[neighbor] == 0 {
+				queue = append(queue, neighbor)
+			}
+		}
+	}
+
+	// Check whether a cycle exists
+	if len(sorted) != len(objs) {
+		return errors.Errorf("There is a circular dependency among the dependencies.")
+	}
+
+	return nil
+}
+
+type DependencyTopoGraph[T any] struct {
+	Graph  map[string][]string
+	Degree map[string]int
+	Leafs  []string // containers whose in-degree is zero
+}
+
+func NewDependencyTopoGraph[T any](
+	objs []T,
+	getId GetObjIdName[T],
+	getName GetObjIdName[T],
+	getDependencies GetDependencies[T],
+) (*DependencyTopoGraph[T], error) {
+	depGraph := &DependencyTopoGraph[T]{
+		Graph:  make(map[string][]string),
+		Degree: make(map[string]int),
+		Leafs:  make([]string, 0, len(objs)),
+	}
+
+	nameToUUID := make(map[string]string)
+
+	for _, obj := range objs {
+		uuid := getId(obj)
+		name := getName(obj)
+
+		depGraph.Degree[uuid] = 0
+		nameToUUID[name] = uuid
+	}
+
+	for _, obj := range objs {
+		for _, dep := range getDependencies(obj) {
+			depId := nameToUUID[dep]
+			uuid := getId(obj)
+			depGraph.Graph[depId] = append(depGraph.Graph[depId], uuid)
+			depGraph.Degree[uuid]++
+		}
+	}
+
+	for uuid, indegree := range depGraph.Degree {
+		if indegree == 0 {
+			depGraph.Leafs = append(depGraph.Leafs, uuid)
+		}
+	}
+
+	return depGraph, nil
+}
+
+type FetchObjById[T any] func(string) T
+
+func (dep *DependencyTopoGraph[T]) GetNextBatch(fetchById FetchObjById[T]) []T {
+	if len(dep.Leafs) == 0 {
+		return nil
+	}
+
+	objs := make([]T, 0, len(dep.Leafs))
+
+	nextLeafs := make([]string, 0)
+
+	for _, uuid := range dep.Leafs {
+		objs = append(objs, fetchById(uuid))
+		for _, neighbor := range dep.Graph[uuid] {
+			dep.Degree[neighbor]--
+			if dep.Degree[neighbor] == 0 {
+				nextLeafs = append(nextLeafs, neighbor)
+			}
+		}
+	}
+
+	// log.Infof("Get next batch: %s", dep.Leafs)
+
+	dep.Leafs = nextLeafs
+
+	return objs
+}

pkg/compute/utils/pod_container_dependencies_test.go

@@ -0,0 +1,129 @@
+package utils
+
+import (
+	"strings"
+	"testing"
+)
+
+type MockContainer struct {
+	ID   string
+	Name string
+	Deps []string
+}
+
+func TestTopologicalSortContainers(t *testing.T) {
+	containers := []MockContainer{
+		{ID: "c1", Name: "container1"},
+		{ID: "c2", Name: "container2", Deps: []string{"container1"}},
+		{ID: "c3", Name: "container3", Deps: []string{"container2"}},
+	}
+
+	mockGetDependencies := func(c MockContainer) []string {
+		return c.Deps
+	}
+
+	err := TopologicalSortContainers(containers, func(c MockContainer) string { return c.Name }, mockGetDependencies)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+}
+
+func TestNewDependencyTopoGraph(t *testing.T) {
+	containers := []MockContainer{
+		{ID: "c1", Name: "container1"},
+		{ID: "c2", Name: "container2", Deps: []string{"container1"}},
+		{ID: "c3", Name: "container3", Deps: []string{"container2"}},
+	}
+
+	mockGetDependencies := func(c MockContainer) []string {
+		return c.Deps
+	}
+
+	graph, err := NewDependencyTopoGraph(
+		containers,
+		func(c MockContainer) string { return c.ID },
+		func(c MockContainer) string { return c.Name },
+		mockGetDependencies,
+	)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	if len(graph.Leafs) != 1 || graph.Leafs[0] != "c1" {
+		t.Errorf("Expected leafs [c1], got %v", graph.Leafs)
+	}
+}
+
+func TestGetNextBatch(t *testing.T) {
+	containers := []MockContainer{
+		{ID: "c1", Name: "container1"},
+		{ID: "c2", Name: "container2", Deps: []string{"container1"}},
+		{ID: "c3", Name: "container3", Deps: []string{"container2"}},
+	}
+
+	mockGetDependencies := func(c MockContainer) []string {
+		return c.Deps
+	}
+
+	mockFetchById := func(id string) MockContainer {
+		for _, c := range containers {
+			if c.ID == id {
+				return c
+			}
+		}
+		return MockContainer{}
+	}
+
+	graph, err := NewDependencyTopoGraph(
+		containers,
+		func(c MockContainer) string { return c.ID },
+		func(c MockContainer) string { return c.Name },
+		mockGetDependencies,
+	)
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	// First batch should be container1
+	batch1 := graph.GetNextBatch(mockFetchById)
+	if len(batch1) != 1 || batch1[0].Name != "container1" {
+		t.Errorf("Expected first batch [container1], got %v", batch1)
+	}
+
+	// Second batch should be container2
+	batch2 := graph.GetNextBatch(mockFetchById)
+	if len(batch2) != 1 || batch2[0].Name != "container2" {
+		t.Errorf("Expected second batch [container2], got %v", batch2)
+	}
+
+	// Third batch should be container3
+	batch3 := graph.GetNextBatch(mockFetchById)
+	if len(batch3) != 1 || batch3[0].Name != "container3" {
+		t.Errorf("Expected third batch [container3], got %v", batch3)
+	}
+
+	// No more batches
+	batch4 := graph.GetNextBatch(mockFetchById)
+	if batch4 != nil {
+		t.Errorf("Expected nil, got %v", batch4)
+	}
+}
+
+func TestCircularDependency(t *testing.T) {
+	containers := []MockContainer{
+		{ID: "c1", Name: "container1", Deps: []string{"container2"}},
+		{ID: "c2", Name: "container2", Deps: []string{"container1"}},
+	}
+
+	mockGetDependencies := func(c MockContainer) []string {
+		return c.Deps
+	}
+
+	err := TopologicalSortContainers(containers, func(c MockContainer) string { return c.Name }, mockGetDependencies)
+	if err == nil {
+		t.Fatal("Expected circular dependency error, got nil")
+	}
+	if !strings.Contains(err.Error(), "circular dependency") {
+		t.Errorf("Expected circular dependency error, got %v", err)
+	}
+}