mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 11:03:34 +00:00
fix: recreate containers only if spec changed by comparing spec hash
This commit is contained in:
@@ -13,6 +13,7 @@ require (
|
||||
github.com/charmbracelet/huh v0.6.0
|
||||
github.com/compose-spec/compose-go/v2 v2.4.5
|
||||
github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
|
||||
github.com/deckarep/golang-set/v2 v2.8.0
|
||||
github.com/dgraph-io/badger/v3 v3.2103.5
|
||||
github.com/distribution/reference v0.6.0
|
||||
github.com/docker/cli v27.5.0+incompatible
|
||||
|
||||
@@ -233,6 +233,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c h1:pFUpOrbxDR6AkioZ1ySsx5yxlDQZ8stG2b88gTPxgJU=
|
||||
github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c/go.mod h1:6UhI8N9EjYm1c2odKpFpAYeR8dsBeM7PtzQhRgxRr9U=
|
||||
github.com/deckarep/golang-set/v2 v2.8.0 h1:swm0rlPCmdWn9mESxKOjWk8hXSqoxOp+ZlfuyaAdFlQ=
|
||||
github.com/deckarep/golang-set/v2 v2.8.0/go.mod h1:VAky9rY/yGXJOLEDv3OMci+7wtDpOF4IN+y82NBOac4=
|
||||
github.com/decred/dcrd/crypto/blake256 v1.0.1 h1:7PltbUIQB7u/FfZ39+DGa/ShuMyJ5ilcvdfma9wOH6Y=
|
||||
github.com/decred/dcrd/crypto/blake256 v1.0.1/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo=
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.3.0 h1:rpfIENRNNilwHwZeG5+P150SMrnNEcHYvcCuK6dPZSg=
|
||||
|
||||
@@ -9,11 +9,12 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
LabelManaged = "uncloud.managed"
|
||||
LabelServiceID = "uncloud.service.id"
|
||||
LabelServiceName = "uncloud.service.name"
|
||||
LabelServiceMode = "uncloud.service.mode"
|
||||
LabelServicePorts = "uncloud.service.ports"
|
||||
LabelManaged = "uncloud.managed"
|
||||
LabelServiceID = "uncloud.service.id"
|
||||
LabelServiceName = "uncloud.service.name"
|
||||
LabelServiceMode = "uncloud.service.mode"
|
||||
LabelServicePorts = "uncloud.service.ports"
|
||||
LabelServiceSpecHash = "uncloud.service.spec-hash"
|
||||
)
|
||||
|
||||
type Container struct {
|
||||
|
||||
+60
-6
@@ -1,6 +1,8 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/distribution/reference"
|
||||
@@ -16,6 +18,12 @@ const (
|
||||
ServiceModeGlobal = "global"
|
||||
)
|
||||
|
||||
var serviceIDRegexp = regexp.MustCompile("^[0-9a-f]{32}$")
|
||||
|
||||
func ValidateServiceID(id string) bool {
|
||||
return serviceIDRegexp.MatchString(id)
|
||||
}
|
||||
|
||||
type ServiceSpec struct {
|
||||
Container ContainerSpec
|
||||
// Mode is the replication mode of the service. Default is ServiceModeReplicated if empty.
|
||||
@@ -50,6 +58,58 @@ func (s *ServiceSpec) Validate() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ImmutableHash returns a hash of the immutable parts of the ServiceSpec that require container recreation if changed.
|
||||
func (s *ServiceSpec) ImmutableHash() (string, error) {
|
||||
var err error
|
||||
// Serialise and sort the ports to ensure the hash is consistent.
|
||||
ports := make([]string, len(s.Ports))
|
||||
for i, p := range s.Ports {
|
||||
ports[i], err = p.String()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("encode service port spec: %w", err)
|
||||
}
|
||||
}
|
||||
slices.Sort(ports)
|
||||
|
||||
volumes := make([]string, 0, len(s.Container.Volumes))
|
||||
volumes = append(volumes, s.Container.Volumes...)
|
||||
slices.Sort(volumes)
|
||||
|
||||
hashSpec := immutableHashSpec{
|
||||
Command: s.Container.Command,
|
||||
Entrypoint: s.Container.Entrypoint,
|
||||
Image: s.Container.Image,
|
||||
Init: s.Container.Init,
|
||||
Ports: ports,
|
||||
Volumes: volumes,
|
||||
}
|
||||
|
||||
data, err := json.Marshal(hashSpec)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("marshal immutable hash spec: %w", err)
|
||||
}
|
||||
|
||||
hasher := sha256.New()
|
||||
if _, err = hasher.Write(data); err != nil {
|
||||
return "", fmt.Errorf("write to SHA256 hasher: %w", err)
|
||||
}
|
||||
|
||||
return hex.EncodeToString(hasher.Sum(nil)), nil
|
||||
}
|
||||
|
||||
// immutableHashSpec contains only the immutable fields from ServiceSpec that require container recreation if changed.
|
||||
type immutableHashSpec struct {
|
||||
Command []string `json:",omitempty"`
|
||||
Entrypoint []string `json:",omitempty"`
|
||||
Image string
|
||||
Init *bool `json:",omitempty"`
|
||||
// Ports are set as labels on the container which are immutable.
|
||||
// TODO: store ingress ports in the cluster store instead of as labels which will allow changing them without
|
||||
// recreating the container.
|
||||
Ports []string `json:",omitempty"`
|
||||
Volumes []string `json:",omitempty"`
|
||||
}
|
||||
|
||||
// Equals returns true if the service spec is equal to the given spec ignoring the number of replicas.
|
||||
func (s *ServiceSpec) Equals(spec ServiceSpec) bool {
|
||||
// TODO: ignore order of ports.
|
||||
@@ -60,12 +120,6 @@ func (s *ServiceSpec) Equals(spec ServiceSpec) bool {
|
||||
return reflect.DeepEqual(*s, spec)
|
||||
}
|
||||
|
||||
var serviceIDRegexp = regexp.MustCompile("^[0-9a-f]{32}$")
|
||||
|
||||
func ValidateServiceID(id string) bool {
|
||||
return serviceIDRegexp.MatchString(id)
|
||||
}
|
||||
|
||||
type ContainerSpec struct {
|
||||
// Command overrides the default CMD of the image to be executed when running a container.
|
||||
Command []string
|
||||
|
||||
@@ -40,17 +40,22 @@ func (cli *Client) CreateContainer(
|
||||
}
|
||||
containerName := fmt.Sprintf("%s-%s", spec.Name, suffix)
|
||||
|
||||
// TODO: calculate the spec hash and set it as a label to detect changes in the service spec.
|
||||
specHash, err := spec.ImmutableHash()
|
||||
if err != nil {
|
||||
return resp, fmt.Errorf("calculate immutable hash for service spec: %w", err)
|
||||
}
|
||||
|
||||
config := &container.Config{
|
||||
Cmd: spec.Container.Command,
|
||||
Entrypoint: spec.Container.Entrypoint,
|
||||
Hostname: containerName,
|
||||
Image: spec.Container.Image,
|
||||
Labels: map[string]string{
|
||||
api.LabelServiceID: serviceID,
|
||||
api.LabelServiceName: spec.Name,
|
||||
api.LabelServiceMode: spec.Mode,
|
||||
api.LabelManaged: "",
|
||||
api.LabelServiceID: serviceID,
|
||||
api.LabelServiceName: spec.Name,
|
||||
api.LabelServiceMode: spec.Mode,
|
||||
api.LabelServiceSpecHash: specHash,
|
||||
api.LabelManaged: "",
|
||||
},
|
||||
}
|
||||
if spec.Mode == "" {
|
||||
@@ -335,3 +340,26 @@ func (cli *Client) RemoveContainer(
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type ContainerSpecStatus string
|
||||
|
||||
const ContainerUpToDate ContainerSpecStatus = "up-to-date"
|
||||
const ContainerNeedsUpdate ContainerSpecStatus = "needs-update"
|
||||
const ContainerNeedsRecreate ContainerSpecStatus = "needs-recreate"
|
||||
|
||||
func CompareContainerToSpec(ctr api.Container, spec api.ServiceSpec) (ContainerSpecStatus, error) {
|
||||
specHash, err := spec.ImmutableHash()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("calculate immutable hash for service spec: %w", err)
|
||||
}
|
||||
|
||||
// Is the hash label is unset, there is no easy way to compare its configuration with the spec,
|
||||
// so let's recreate as well.
|
||||
if ctr.Config.Labels[api.LabelServiceSpecHash] != specHash {
|
||||
return ContainerNeedsRecreate, nil
|
||||
}
|
||||
|
||||
// TODO: compare mutable properties such as memory or CPU limits when they are implemented.
|
||||
|
||||
return ContainerUpToDate, nil
|
||||
}
|
||||
|
||||
@@ -90,30 +90,31 @@ func (s *RollingStrategy) planReplicated(
|
||||
// Organise existing containers by machine.
|
||||
containersOnMachine := make(map[string][]api.Container)
|
||||
upToDateContainersOnMachine := make(map[string]int)
|
||||
containerSpecStatuses := make(map[string]ContainerSpecStatus)
|
||||
if svc != nil {
|
||||
runningSpecs := make(map[string]api.ServiceSpec)
|
||||
for _, c := range svc.Containers {
|
||||
if !c.Container.State.Running || c.Container.State.Paused {
|
||||
// Skip containers that are not running.
|
||||
continue
|
||||
}
|
||||
// TODO: determine if the spec has changed by comparing the hashes.
|
||||
// Refactor all the spec comparison logic below.
|
||||
cs, err := c.Container.ServiceSpec()
|
||||
if err == nil {
|
||||
runningSpecs[c.Container.ID] = cs
|
||||
if cs.Equals(spec) {
|
||||
upToDateContainersOnMachine[c.MachineID] += 1
|
||||
}
|
||||
|
||||
status, err := CompareContainerToSpec(c.Container, spec)
|
||||
if err != nil {
|
||||
return plan, fmt.Errorf("compare container to spec: %w", err)
|
||||
}
|
||||
containerSpecStatuses[c.Container.ID] = status
|
||||
|
||||
if status == ContainerUpToDate {
|
||||
upToDateContainersOnMachine[c.MachineID] += 1
|
||||
}
|
||||
}
|
||||
|
||||
// Sort containers such that containers with the desired spec are first.
|
||||
// Sort containers such that running containers with the desired spec are first.
|
||||
slices.SortFunc(svc.Containers, func(c1, c2 api.MachineContainer) int {
|
||||
if spec1, ok := runningSpecs[c1.Container.ID]; ok && spec1.Equals(spec) {
|
||||
if status, ok := containerSpecStatuses[c1.Container.ID]; ok && status == ContainerUpToDate {
|
||||
return -1
|
||||
}
|
||||
if spec2, ok := runningSpecs[c2.Container.ID]; ok && spec2.Equals(spec) {
|
||||
if status, ok := containerSpecStatuses[c2.Container.ID]; ok && status == ContainerUpToDate {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
@@ -158,14 +159,14 @@ func (s *RollingStrategy) planReplicated(
|
||||
ctr := containers[0]
|
||||
containersOnMachine[m.Id] = containers[1:]
|
||||
|
||||
if ctr.State.Running {
|
||||
ctrSpec, specErr := ctr.ServiceSpec()
|
||||
if specErr == nil && ctrSpec.Equals(spec) {
|
||||
if status, ok := containerSpecStatuses[ctr.ID]; ok { // Contains statuses for only running containers.
|
||||
if status == ContainerUpToDate {
|
||||
continue
|
||||
}
|
||||
// TODO: handle ContainerNeedsUpdate when update of mutable fields on a container is supported.
|
||||
|
||||
conflictingPorts, portsErr := ctr.ConflictingServicePorts(spec.Ports)
|
||||
if specErr != nil || portsErr != nil || len(conflictingPorts) > 0 {
|
||||
if portsErr != nil || len(conflictingPorts) > 0 {
|
||||
// Stop the malformed container or the container with conflicting ports.
|
||||
plan.Operations = append(plan.Operations, &StopContainerOperation{
|
||||
ServiceID: plan.ServiceID,
|
||||
@@ -245,12 +246,12 @@ func (s *RollingStrategy) planGlobal(
|
||||
}
|
||||
|
||||
// TODO: figure out how to return a warning if there are machines down. Embed the machinesDown in the plan?
|
||||
// WARNING: failed to run a service container on machine '%s' which is Down.
|
||||
var machinesDown []*pb.MachineInfo
|
||||
for _, m := range machines {
|
||||
// Skip machines that are down but collect them to report a warning later.
|
||||
if m.State == pb.MachineMember_DOWN {
|
||||
machinesDown = append(machinesDown, m.Machine)
|
||||
fmt.Printf("WARNING: failed to run a service container on machine '%s' which is Down.\n", m.Machine.Id)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -291,11 +292,11 @@ func reconcileGlobalContainer(
|
||||
continue
|
||||
}
|
||||
|
||||
svcSpec, err := c.Container.ServiceSpec()
|
||||
status, err := CompareContainerToSpec(c.Container, spec)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get service spec: %w", err)
|
||||
return nil, fmt.Errorf("compare container to spec: %w", err)
|
||||
}
|
||||
if svcSpec.Equals(spec) {
|
||||
if status == ContainerUpToDate {
|
||||
// The container is already running with the same spec.
|
||||
upToDate = true
|
||||
for j, old := range containers {
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package e2e
|
||||
|
||||
import (
|
||||
mapset "github.com/deckarep/golang-set/v2"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"testing"
|
||||
"uncloud/internal/api"
|
||||
"uncloud/internal/cli/client"
|
||||
)
|
||||
|
||||
func assertServiceMatchesSpec(t *testing.T, svc api.Service, spec api.ServiceSpec) {
|
||||
assert.Equal(t, spec.Name, svc.Name)
|
||||
assert.Equal(t, spec.Mode, svc.Mode)
|
||||
|
||||
if svc.Mode == api.ServiceModeReplicated {
|
||||
assert.Len(t, svc.Containers, int(spec.Replicas), "Expected %d replicas", spec.Replicas)
|
||||
}
|
||||
|
||||
for _, mc := range svc.Containers {
|
||||
assertContainerMatchesSpec(t, mc.Container, spec)
|
||||
}
|
||||
}
|
||||
|
||||
func assertContainerMatchesSpec(t *testing.T, ctr api.Container, spec api.ServiceSpec) {
|
||||
status, err := client.CompareContainerToSpec(ctr, spec)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, client.ContainerUpToDate, status)
|
||||
}
|
||||
|
||||
// serviceContainersByMachine returns a map of machine ID to service containers on that machine.
|
||||
func serviceContainersByMachine(t *testing.T, svc api.Service) map[string][]api.Container {
|
||||
containers := make(map[string][]api.Container)
|
||||
for _, c := range svc.Containers {
|
||||
containers[c.MachineID] = append(containers[c.MachineID], c.Container)
|
||||
}
|
||||
return containers
|
||||
}
|
||||
|
||||
func serviceMachines(t *testing.T, svc api.Service) mapset.Set[string] {
|
||||
machines := mapset.NewSet[string]()
|
||||
for _, c := range svc.Containers {
|
||||
machines.Add(c.MachineID)
|
||||
}
|
||||
|
||||
return machines
|
||||
}
|
||||
|
||||
func serviceContainerIDs(t *testing.T, svc api.Service) mapset.Set[string] {
|
||||
ids := mapset.NewSet[string]()
|
||||
for _, c := range svc.Containers {
|
||||
ids.Add(c.Container.ID)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
+107
-104
@@ -66,7 +66,7 @@ func TestDeployment(t *testing.T) {
|
||||
plan, err := deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.NotEmpty(t, plan.ServiceID)
|
||||
assert.NotEmpty(t, plan.ServiceName)
|
||||
assert.Equal(t, name, plan.ServiceName)
|
||||
assert.Len(t, plan.SequenceOperation.Operations, 3) // 3 run
|
||||
|
||||
runPlan, err := deploy.Run(ctx)
|
||||
@@ -75,21 +75,15 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
svc, err := cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
|
||||
assertServiceMatchesSpec(t, svc, spec)
|
||||
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
|
||||
svcSpec, err := svc.Containers[0].Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(spec))
|
||||
|
||||
machines := make(map[string]struct{})
|
||||
for _, ctr := range svc.Containers {
|
||||
machines[ctr.MachineID] = struct{}{}
|
||||
}
|
||||
assert.Len(t, machines, 3, "expected 1 container on each machine")
|
||||
machines := serviceMachines(t, svc)
|
||||
assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine")
|
||||
|
||||
// Deploy a published port.
|
||||
initialContainers := serviceContainerIDs(t, svc)
|
||||
|
||||
specWithPort := api.ServiceSpec{
|
||||
Name: name,
|
||||
Mode: api.ServiceModeGlobal,
|
||||
@@ -117,15 +111,18 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
assertServiceMatchesSpec(t, svc, specWithPort)
|
||||
|
||||
svcSpec, err = svc.Containers[0].Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(specWithPort))
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
machines = serviceMachines(t, svc)
|
||||
assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine")
|
||||
containers := serviceContainerIDs(t, svc)
|
||||
assert.Empty(t, initialContainers.Intersect(containers).ToSlice(),
|
||||
"All existing containers should be replaced")
|
||||
|
||||
// Deploy the same conflicting port but with container spec changes
|
||||
initialContainers = containers
|
||||
|
||||
init := true
|
||||
specWithPortAndInit := api.ServiceSpec{
|
||||
Name: name,
|
||||
@@ -155,15 +152,18 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
assertServiceMatchesSpec(t, svc, specWithPortAndInit)
|
||||
|
||||
svcSpec, err = svc.Containers[0].Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(specWithPortAndInit))
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
machines = serviceMachines(t, svc)
|
||||
assert.Len(t, machines.ToSlice(), 3, "Expected 1 container on each machine")
|
||||
containers = serviceContainerIDs(t, svc)
|
||||
assert.Empty(t, initialContainers.Intersect(containers).ToSlice(),
|
||||
"All existing containers should be replaced")
|
||||
|
||||
// Deploying the same spec should be a no-op.
|
||||
initialContainers = containers
|
||||
|
||||
deploy, err = cli.NewDeployment(specWithPortAndInit, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -176,9 +176,9 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Equal(t, api.ServiceModeGlobal, svc.Mode)
|
||||
assert.Len(t, svc.Containers, 3)
|
||||
|
||||
containers = serviceContainerIDs(t, svc)
|
||||
assert.ElementsMatch(t, initialContainers.ToSlice(), containers.ToSlice())
|
||||
})
|
||||
|
||||
t.Run("global with machine filter", func(t *testing.T) {
|
||||
@@ -432,6 +432,8 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
plan, err := deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.NotEmpty(t, plan.ServiceID)
|
||||
assert.Equal(t, name, plan.ServiceName)
|
||||
assert.Len(t, plan.SequenceOperation.Operations, 2) // 2 run operations for 2 replicas
|
||||
|
||||
runPlan, err := deploy.Run(ctx)
|
||||
@@ -441,27 +443,12 @@ func TestDeployment(t *testing.T) {
|
||||
// Verify service was created with correct settings.
|
||||
svc, err := cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Equal(t, api.ServiceModeReplicated, svc.Mode)
|
||||
assert.Len(t, svc.Containers, 2, "expected 2 replicas")
|
||||
assertServiceMatchesSpec(t, svc, spec)
|
||||
|
||||
// Verify containers are on different machines for balanced distribution.
|
||||
machines := make(map[string]struct{})
|
||||
for _, ctr := range svc.Containers {
|
||||
machines[ctr.MachineID] = struct{}{}
|
||||
|
||||
// Verify container spec matches our deployment spec.
|
||||
svcSpec, err := ctr.Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(spec))
|
||||
}
|
||||
assert.Len(t, machines, 2, "containers should be on different machines")
|
||||
|
||||
// Store the initial container IDs.
|
||||
initialContainers := make(map[string]string) // machineID -> containerID
|
||||
for _, ctr := range svc.Containers {
|
||||
initialContainers[ctr.MachineID] = ctr.Container.ID
|
||||
}
|
||||
initialMachines := serviceMachines(t, svc)
|
||||
assert.Len(t, initialMachines.ToSlice(), 2, "Expected 2 containers on 2 different machines")
|
||||
initialContainers := serviceContainerIDs(t, svc)
|
||||
|
||||
// 2. Update the service with a new configuration.
|
||||
init := true
|
||||
@@ -473,90 +460,101 @@ func TestDeployment(t *testing.T) {
|
||||
|
||||
plan, err = deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, plan.Operations, 4, "expected 2 run + 2 remove operations")
|
||||
assert.Len(t, plan.Operations, 4, "Expected 2 run + 2 remove operations")
|
||||
|
||||
_, err = deploy.Run(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify service was updated.
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Len(t, svc.Containers, 2)
|
||||
assertServiceMatchesSpec(t, svc, updatedSpec)
|
||||
|
||||
// Verify initial containers were updated.
|
||||
for _, ctr := range svc.Containers {
|
||||
initialCtr, ok := initialContainers[ctr.MachineID]
|
||||
require.True(t, ok, "Updated container should have replaced one of the initial containers")
|
||||
// Verify containers are on the same machines as before but the initial containers were replaced.
|
||||
machines := serviceMachines(t, svc)
|
||||
assert.ElementsMatch(t, initialMachines.ToSlice(), machines.ToSlice(),
|
||||
"Expected containers on the same machines")
|
||||
containers := serviceContainerIDs(t, svc)
|
||||
assert.Empty(t, initialContainers.Intersect(containers).ToSlice(),
|
||||
"All existing containers should be replaced")
|
||||
|
||||
assert.NotEqual(t, initialCtr, ctr.Container.ID,
|
||||
"Container on machine %s should have been updated", ctr.MachineID)
|
||||
// 3. Scale to 3 replicas.
|
||||
initialMachines = machines
|
||||
initialContainers = containers // Reset container tracking.
|
||||
|
||||
svcSpec, err := ctr.Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(updatedSpec))
|
||||
}
|
||||
threeReplicaSpec := updatedSpec
|
||||
threeReplicaSpec.Replicas = 3
|
||||
|
||||
// 3. Update to 4 replicas with a different configuration.
|
||||
initialContainers = make(map[string]string) // Reset container tracking.
|
||||
for _, ctr := range svc.Containers {
|
||||
initialContainers[ctr.MachineID] = ctr.Container.ID
|
||||
}
|
||||
deploy, err = cli.NewDeployment(threeReplicaSpec, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
plan, err = deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, plan.Operations, 1, "Expected 1 run operation")
|
||||
|
||||
_, err = deploy.Run(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assertServiceMatchesSpec(t, svc, threeReplicaSpec)
|
||||
|
||||
// Verify existing containers remain and a new one was added on a different machine.
|
||||
machines = serviceMachines(t, svc)
|
||||
assert.Len(t, machines.ToSlice(), 3, "Expected 3 containers on 3 different machines")
|
||||
containers = serviceContainerIDs(t, svc)
|
||||
assert.Len(t, containers.Intersect(initialContainers).ToSlice(), 2, "Expected 2 initial containers to remain")
|
||||
|
||||
// 4. Update to 5 replicas with a different configuration.
|
||||
initialContainers = containers // Reset container tracking.
|
||||
|
||||
fourReplicaSpec := updatedSpec
|
||||
fourReplicaSpec.Container.Command = []string{"updated"}
|
||||
fourReplicaSpec.Replicas = 4
|
||||
fourReplicaSpec.Replicas = 5
|
||||
|
||||
deploy, err = cli.NewDeployment(fourReplicaSpec, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
plan, err = deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.Len(t, plan.Operations, 6, "Expected 4 run + 2 remove operations")
|
||||
assert.Len(t, plan.Operations, 8, "Expected 5 run + 3 remove operations")
|
||||
|
||||
_, err = deploy.Run(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Verify service now has 4 containers.
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, name, svc.Name)
|
||||
assert.Len(t, svc.Containers, 4, "Expected 4 replicas")
|
||||
assertServiceMatchesSpec(t, svc, fourReplicaSpec)
|
||||
|
||||
// Count containers per machine.
|
||||
machineContainerCount := make(map[string]int)
|
||||
for _, ctr := range svc.Containers {
|
||||
machineContainerCount[ctr.MachineID]++
|
||||
// Verify all existing containers were replaced and new ones are evenly distributed.
|
||||
machines = serviceMachines(t, svc)
|
||||
assert.Len(t, machines.ToSlice(), 3, "Expected containers on 3 different machines")
|
||||
containers = serviceContainerIDs(t, svc)
|
||||
assert.Empty(t, containers.Intersect(initialContainers).ToSlice(),
|
||||
"All existing containers should be replaced")
|
||||
|
||||
// Verify all containers match the new spec
|
||||
svcSpec, err := ctr.Container.ServiceSpec()
|
||||
require.NoError(t, err)
|
||||
assert.True(t, svcSpec.Equals(fourReplicaSpec))
|
||||
|
||||
// For existing machines, verify containers were replaced
|
||||
if initialID, ok := initialContainers[ctr.MachineID]; ok {
|
||||
assert.NotEqual(t, initialID, ctr.Container.ID,
|
||||
"Container on machine %s should have been updated", ctr.MachineID)
|
||||
}
|
||||
machineContainers := serviceContainersByMachine(t, svc)
|
||||
for _, ctrs := range machineContainers {
|
||||
assert.LessOrEqual(t, len(ctrs), 2, "Expected at most 2 containers on each machine")
|
||||
}
|
||||
|
||||
// Verify even distributions across machines.
|
||||
assert.Len(t, machineContainerCount, 3, "Expected containers on all 3 machines")
|
||||
for _, count := range machineContainerCount {
|
||||
assert.GreaterOrEqual(t, count, 1, "Expected at least 1 container on each machine")
|
||||
}
|
||||
// 5. Redeploy the exact same spec and verify it's a noop.
|
||||
initialContainers = containers // Reset container tracking.
|
||||
|
||||
// 4. Redeploy the exact same spec and verify it's a noop.
|
||||
deploy, err = cli.NewDeployment(fourReplicaSpec, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
plan, err = deploy.Plan(ctx)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, plan.Operations, "Redeploying the same spec should be a no-op")
|
||||
|
||||
_, err = deploy.Run(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
svc, err = cli.InspectService(ctx, name)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Empty(t, plan.Operations, "Redeploying the same spec should be a no-op")
|
||||
containers = serviceContainerIDs(t, svc)
|
||||
assert.ElementsMatch(t, initialContainers.ToSlice(), containers.ToSlice())
|
||||
})
|
||||
|
||||
t.Run("replicated with machine filter", func(t *testing.T) {
|
||||
@@ -637,10 +635,10 @@ func TestDeployment(t *testing.T) {
|
||||
// TODO: test deployments with unreachable machines. See https://github.com/psviderski/uncloud/issues/29.
|
||||
}
|
||||
|
||||
func TestRunService(t *testing.T) {
|
||||
func TestServiceLifecycle(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
clusterName := "ucind-test.run-service"
|
||||
clusterName := "ucind-test.service"
|
||||
ctx := context.Background()
|
||||
c, _ := createTestCluster(t, clusterName, ucind.CreateClusterOptions{Machines: 3}, true)
|
||||
|
||||
@@ -680,6 +678,7 @@ func TestRunService(t *testing.T) {
|
||||
|
||||
// Verify default settings.
|
||||
assert.Empty(t, ctr.Config.Cmd)
|
||||
assert.EqualValues(t, []string{"/pause"}, ctr.Config.Entrypoint) // Populated by the image.
|
||||
assert.Nil(t, ctr.HostConfig.Init)
|
||||
assert.Empty(t, ctr.HostConfig.Binds)
|
||||
assert.Empty(t, ctr.HostConfig.PortBindings)
|
||||
@@ -710,9 +709,11 @@ func TestRunService(t *testing.T) {
|
||||
Mode: api.ServiceModeGlobal,
|
||||
Container: api.ContainerSpec{
|
||||
Command: []string{"sleep", "infinity"},
|
||||
Image: "portainer/pause:latest",
|
||||
Init: &init,
|
||||
Volumes: []string{"/host/path:/container/path:ro"},
|
||||
// Extra slashes is not a typo, it changes the spec but Linux ignores them and uses the default /pause.
|
||||
Entrypoint: []string{"///pause"},
|
||||
Image: "portainer/pause:latest",
|
||||
Init: &init,
|
||||
Volumes: []string{"/host/path:/container/path:ro"},
|
||||
},
|
||||
Ports: []api.PortSpec{
|
||||
{
|
||||
@@ -751,6 +752,7 @@ func TestRunService(t *testing.T) {
|
||||
assert.Equal(t, "portainer/pause:latest", ctr.Config.Image)
|
||||
|
||||
assert.EqualValues(t, spec.Container.Command, ctr.Config.Cmd)
|
||||
assert.EqualValues(t, spec.Container.Entrypoint, ctr.Config.Entrypoint)
|
||||
assert.True(t, *ctr.HostConfig.Init)
|
||||
assert.Len(t, ctr.HostConfig.Binds, 1)
|
||||
assert.Contains(t, ctr.HostConfig.Binds, spec.Container.Volumes[0])
|
||||
@@ -894,16 +896,17 @@ func TestRunService(t *testing.T) {
|
||||
Protocol: api.ProtocolHTTPS,
|
||||
Mode: api.PortModeIngress,
|
||||
},
|
||||
{
|
||||
PublishedPort: 8000,
|
||||
ContainerPort: 8080,
|
||||
Protocol: api.ProtocolTCP,
|
||||
Mode: api.PortModeIngress,
|
||||
},
|
||||
// Not supported yet.
|
||||
//{
|
||||
// PublishedPort: 8000,
|
||||
// ContainerPort: 8080,
|
||||
// Protocol: api.ProtocolTCP,
|
||||
// Mode: api.PortModeIngress,
|
||||
//},
|
||||
{
|
||||
PublishedPort: 8000,
|
||||
ContainerPort: 8000,
|
||||
Protocol: api.ProtocolUDP,
|
||||
Protocol: api.ProtocolTCP,
|
||||
Mode: api.PortModeHost,
|
||||
},
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user