convergence.go 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938
  1. /*
  2. Copyright 2020 Docker Compose CLI authors
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package compose
  14. import (
  15. "context"
  16. "errors"
  17. "fmt"
  18. "maps"
  19. "slices"
  20. "sort"
  21. "strconv"
  22. "strings"
  23. "sync"
  24. "time"
  25. "github.com/compose-spec/compose-go/v2/types"
  26. "github.com/containerd/platforms"
  27. containerType "github.com/docker/docker/api/types/container"
  28. mmount "github.com/docker/docker/api/types/mount"
  29. "github.com/docker/docker/api/types/versions"
  30. specs "github.com/opencontainers/image-spec/specs-go/v1"
  31. "github.com/sirupsen/logrus"
  32. "go.opentelemetry.io/otel/attribute"
  33. "go.opentelemetry.io/otel/trace"
  34. "golang.org/x/sync/errgroup"
  35. "github.com/docker/compose/v2/internal/tracing"
  36. "github.com/docker/compose/v2/pkg/api"
  37. "github.com/docker/compose/v2/pkg/progress"
  38. "github.com/docker/compose/v2/pkg/utils"
  39. )
  40. const (
  41. doubledContainerNameWarning = "WARNING: The %q service is using the custom container name %q. " +
  42. "Docker requires each container to have a unique name. " +
  43. "Remove the custom name to scale the service"
  44. )
  45. // convergence manages service's container lifecycle.
  46. // Based on initially observed state, it reconciles the existing container with desired state, which might include
  47. // re-creating container, adding or removing replicas, or starting stopped containers.
  48. // Cross services dependencies are managed by creating services in expected order and updating `service:xx` reference
  49. // when a service has converged, so dependent ones can be managed with resolved containers references.
  50. type convergence struct {
  51. service *composeService
  52. services map[string]Containers
  53. networks map[string]string
  54. volumes map[string]string
  55. stateMutex sync.Mutex
  56. }
  57. func (c *convergence) getObservedState(serviceName string) Containers {
  58. c.stateMutex.Lock()
  59. defer c.stateMutex.Unlock()
  60. return c.services[serviceName]
  61. }
  62. func (c *convergence) setObservedState(serviceName string, containers Containers) {
  63. c.stateMutex.Lock()
  64. defer c.stateMutex.Unlock()
  65. c.services[serviceName] = containers
  66. }
  67. func newConvergence(services []string, state Containers, networks map[string]string, volumes map[string]string, s *composeService) *convergence {
  68. observedState := map[string]Containers{}
  69. for _, s := range services {
  70. observedState[s] = Containers{}
  71. }
  72. for _, c := range state.filter(isNotOneOff) {
  73. service := c.Labels[api.ServiceLabel]
  74. observedState[service] = append(observedState[service], c)
  75. }
  76. return &convergence{
  77. service: s,
  78. services: observedState,
  79. networks: networks,
  80. volumes: volumes,
  81. }
  82. }
  83. func (c *convergence) apply(ctx context.Context, project *types.Project, options api.CreateOptions) error {
  84. return InDependencyOrder(ctx, project, func(ctx context.Context, name string) error {
  85. service, err := project.GetService(name)
  86. if err != nil {
  87. return err
  88. }
  89. return tracing.SpanWrapFunc("service/apply", tracing.ServiceOptions(service), func(ctx context.Context) error {
  90. strategy := options.RecreateDependencies
  91. if slices.Contains(options.Services, name) {
  92. strategy = options.Recreate
  93. }
  94. return c.ensureService(ctx, project, service, strategy, options.Inherit, options.Timeout)
  95. })(ctx)
  96. })
  97. }
  98. func (c *convergence) ensureService(ctx context.Context, project *types.Project, service types.ServiceConfig, recreate string, inherit bool, timeout *time.Duration) error { //nolint:gocyclo
  99. if service.Provider != nil {
  100. return c.service.runPlugin(ctx, project, service, "up")
  101. }
  102. expected, err := getScale(service)
  103. if err != nil {
  104. return err
  105. }
  106. containers := c.getObservedState(service.Name)
  107. actual := len(containers)
  108. updated := make(Containers, expected)
  109. eg, _ := errgroup.WithContext(ctx)
  110. err = c.resolveServiceReferences(&service)
  111. if err != nil {
  112. return err
  113. }
  114. sort.Slice(containers, func(i, j int) bool {
  115. // select obsolete containers first, so they get removed as we scale down
  116. if obsolete, _ := c.mustRecreate(service, containers[i], recreate); obsolete {
  117. // i is obsolete, so must be first in the list
  118. return true
  119. }
  120. if obsolete, _ := c.mustRecreate(service, containers[j], recreate); obsolete {
  121. // j is obsolete, so must be first in the list
  122. return false
  123. }
  124. // For up-to-date containers, sort by container number to preserve low-values in container numbers
  125. ni, erri := strconv.Atoi(containers[i].Labels[api.ContainerNumberLabel])
  126. nj, errj := strconv.Atoi(containers[j].Labels[api.ContainerNumberLabel])
  127. if erri == nil && errj == nil {
  128. return ni > nj
  129. }
  130. // If we don't get a container number (?) just sort by creation date
  131. return containers[i].Created < containers[j].Created
  132. })
  133. slices.Reverse(containers)
  134. for i, container := range containers {
  135. if i >= expected {
  136. // Scale Down
  137. // As we sorted containers, obsolete ones and/or highest number will be removed
  138. container := container
  139. traceOpts := append(tracing.ServiceOptions(service), tracing.ContainerOptions(container)...)
  140. eg.Go(tracing.SpanWrapFuncForErrGroup(ctx, "service/scale/down", traceOpts, func(ctx context.Context) error {
  141. return c.service.stopAndRemoveContainer(ctx, container, &service, timeout, false)
  142. }))
  143. continue
  144. }
  145. mustRecreate, err := c.mustRecreate(service, container, recreate)
  146. if err != nil {
  147. return err
  148. }
  149. if mustRecreate {
  150. err := c.stopDependentContainers(ctx, project, service)
  151. if err != nil {
  152. return err
  153. }
  154. i, container := i, container
  155. eg.Go(tracing.SpanWrapFuncForErrGroup(ctx, "container/recreate", tracing.ContainerOptions(container), func(ctx context.Context) error {
  156. recreated, err := c.service.recreateContainer(ctx, project, service, container, inherit, timeout)
  157. updated[i] = recreated
  158. return err
  159. }))
  160. continue
  161. }
  162. // Enforce non-diverged containers are running
  163. w := progress.ContextWriter(ctx)
  164. name := getContainerProgressName(container)
  165. switch container.State {
  166. case ContainerRunning:
  167. w.Event(progress.RunningEvent(name))
  168. case ContainerCreated:
  169. case ContainerRestarting:
  170. case ContainerExited:
  171. default:
  172. container := container
  173. eg.Go(tracing.EventWrapFuncForErrGroup(ctx, "service/start", tracing.ContainerOptions(container), func(ctx context.Context) error {
  174. return c.service.startContainer(ctx, container)
  175. }))
  176. }
  177. updated[i] = container
  178. }
  179. next := nextContainerNumber(containers)
  180. for i := 0; i < expected-actual; i++ {
  181. // Scale UP
  182. number := next + i
  183. name := getContainerName(project.Name, service, number)
  184. eventOpts := tracing.SpanOptions{trace.WithAttributes(attribute.String("container.name", name))}
  185. eg.Go(tracing.EventWrapFuncForErrGroup(ctx, "service/scale/up", eventOpts, func(ctx context.Context) error {
  186. opts := createOptions{
  187. AutoRemove: false,
  188. AttachStdin: false,
  189. UseNetworkAliases: true,
  190. Labels: mergeLabels(service.Labels, service.CustomLabels),
  191. }
  192. container, err := c.service.createContainer(ctx, project, service, name, number, opts)
  193. updated[actual+i] = container
  194. return err
  195. }))
  196. continue
  197. }
  198. err = eg.Wait()
  199. c.setObservedState(service.Name, updated)
  200. return err
  201. }
  202. func (c *convergence) stopDependentContainers(ctx context.Context, project *types.Project, service types.ServiceConfig) error {
  203. // Stop dependent containers, so they will be restarted after service is re-created
  204. dependents := project.GetDependentsForService(service, func(dependency types.ServiceDependency) bool {
  205. return dependency.Restart
  206. })
  207. if len(dependents) == 0 {
  208. return nil
  209. }
  210. err := c.service.stop(ctx, project.Name, api.StopOptions{
  211. Services: dependents,
  212. Project: project,
  213. }, nil)
  214. if err != nil {
  215. return err
  216. }
  217. for _, name := range dependents {
  218. dependentStates := c.getObservedState(name)
  219. for i, dependent := range dependentStates {
  220. dependent.State = ContainerExited
  221. dependentStates[i] = dependent
  222. }
  223. c.setObservedState(name, dependentStates)
  224. }
  225. return nil
  226. }
  227. func getScale(config types.ServiceConfig) (int, error) {
  228. scale := config.GetScale()
  229. if scale > 1 && config.ContainerName != "" {
  230. return 0, fmt.Errorf(doubledContainerNameWarning,
  231. config.Name,
  232. config.ContainerName)
  233. }
  234. return scale, nil
  235. }
  236. // resolveServiceReferences replaces reference to another service with reference to an actual container
  237. func (c *convergence) resolveServiceReferences(service *types.ServiceConfig) error {
  238. err := c.resolveVolumeFrom(service)
  239. if err != nil {
  240. return err
  241. }
  242. err = c.resolveSharedNamespaces(service)
  243. if err != nil {
  244. return err
  245. }
  246. return nil
  247. }
  248. func (c *convergence) resolveVolumeFrom(service *types.ServiceConfig) error {
  249. for i, vol := range service.VolumesFrom {
  250. spec := strings.Split(vol, ":")
  251. if len(spec) == 0 {
  252. continue
  253. }
  254. if spec[0] == "container" {
  255. service.VolumesFrom[i] = spec[1]
  256. continue
  257. }
  258. name := spec[0]
  259. dependencies := c.getObservedState(name)
  260. if len(dependencies) == 0 {
  261. return fmt.Errorf("cannot share volume with service %s: container missing", name)
  262. }
  263. service.VolumesFrom[i] = dependencies.sorted()[0].ID
  264. }
  265. return nil
  266. }
  267. func (c *convergence) resolveSharedNamespaces(service *types.ServiceConfig) error {
  268. str := service.NetworkMode
  269. if name := getDependentServiceFromMode(str); name != "" {
  270. dependencies := c.getObservedState(name)
  271. if len(dependencies) == 0 {
  272. return fmt.Errorf("cannot share network namespace with service %s: container missing", name)
  273. }
  274. service.NetworkMode = types.ContainerPrefix + dependencies.sorted()[0].ID
  275. }
  276. str = service.Ipc
  277. if name := getDependentServiceFromMode(str); name != "" {
  278. dependencies := c.getObservedState(name)
  279. if len(dependencies) == 0 {
  280. return fmt.Errorf("cannot share IPC namespace with service %s: container missing", name)
  281. }
  282. service.Ipc = types.ContainerPrefix + dependencies.sorted()[0].ID
  283. }
  284. str = service.Pid
  285. if name := getDependentServiceFromMode(str); name != "" {
  286. dependencies := c.getObservedState(name)
  287. if len(dependencies) == 0 {
  288. return fmt.Errorf("cannot share PID namespace with service %s: container missing", name)
  289. }
  290. service.Pid = types.ContainerPrefix + dependencies.sorted()[0].ID
  291. }
  292. return nil
  293. }
  294. func (c *convergence) mustRecreate(expected types.ServiceConfig, actual containerType.Summary, policy string) (bool, error) {
  295. if policy == api.RecreateNever {
  296. return false, nil
  297. }
  298. if policy == api.RecreateForce {
  299. return true, nil
  300. }
  301. configHash, err := ServiceHash(expected)
  302. if err != nil {
  303. return false, err
  304. }
  305. configChanged := actual.Labels[api.ConfigHashLabel] != configHash
  306. imageUpdated := actual.Labels[api.ImageDigestLabel] != expected.CustomLabels[api.ImageDigestLabel]
  307. if configChanged || imageUpdated {
  308. return true, nil
  309. }
  310. if c.networks != nil && actual.State == "running" {
  311. if checkExpectedNetworks(expected, actual, c.networks) {
  312. return true, nil
  313. }
  314. }
  315. if c.volumes != nil {
  316. if checkExpectedVolumes(expected, actual, c.volumes) {
  317. return true, nil
  318. }
  319. }
  320. return false, nil
  321. }
  322. func checkExpectedNetworks(expected types.ServiceConfig, actual containerType.Summary, networks map[string]string) bool {
  323. // check the networks container is connected to are the expected ones
  324. for net := range expected.Networks {
  325. id := networks[net]
  326. if id == "swarm" {
  327. // corner-case : swarm overlay network isn't visible until a container is attached
  328. continue
  329. }
  330. found := false
  331. for _, settings := range actual.NetworkSettings.Networks {
  332. if settings.NetworkID == id {
  333. found = true
  334. break
  335. }
  336. }
  337. if !found {
  338. // config is up-to-date but container is not connected to network
  339. return true
  340. }
  341. }
  342. return false
  343. }
  344. func checkExpectedVolumes(expected types.ServiceConfig, actual containerType.Summary, volumes map[string]string) bool {
  345. // check container's volume mounts and search for the expected ones
  346. for _, vol := range expected.Volumes {
  347. if vol.Type != string(mmount.TypeVolume) {
  348. continue
  349. }
  350. if vol.Source == "" {
  351. continue
  352. }
  353. id := volumes[vol.Source]
  354. found := false
  355. for _, mount := range actual.Mounts {
  356. if mount.Type != mmount.TypeVolume {
  357. continue
  358. }
  359. if mount.Name == id {
  360. found = true
  361. break
  362. }
  363. }
  364. if !found {
  365. // config is up-to-date but container doesn't have volume mounted
  366. return true
  367. }
  368. }
  369. return false
  370. }
  371. func getContainerName(projectName string, service types.ServiceConfig, number int) string {
  372. name := getDefaultContainerName(projectName, service.Name, strconv.Itoa(number))
  373. if service.ContainerName != "" {
  374. name = service.ContainerName
  375. }
  376. return name
  377. }
  378. func getDefaultContainerName(projectName, serviceName, index string) string {
  379. return strings.Join([]string{projectName, serviceName, index}, api.Separator)
  380. }
  381. func getContainerProgressName(ctr containerType.Summary) string {
  382. return "Container " + getCanonicalContainerName(ctr)
  383. }
  384. func containerEvents(containers Containers, eventFunc func(string) progress.Event) []progress.Event {
  385. events := []progress.Event{}
  386. for _, ctr := range containers {
  387. events = append(events, eventFunc(getContainerProgressName(ctr)))
  388. }
  389. return events
  390. }
  391. func containerReasonEvents(containers Containers, eventFunc func(string, string) progress.Event, reason string) []progress.Event {
  392. events := []progress.Event{}
  393. for _, ctr := range containers {
  394. events = append(events, eventFunc(getContainerProgressName(ctr), reason))
  395. }
  396. return events
  397. }
  398. // ServiceConditionRunningOrHealthy is a service condition on status running or healthy
  399. const ServiceConditionRunningOrHealthy = "running_or_healthy"
  400. //nolint:gocyclo
  401. func (s *composeService) waitDependencies(ctx context.Context, project *types.Project, dependant string, dependencies types.DependsOnConfig, containers Containers, timeout time.Duration) error {
  402. if timeout > 0 {
  403. withTimeout, cancelFunc := context.WithTimeout(ctx, timeout)
  404. defer cancelFunc()
  405. ctx = withTimeout
  406. }
  407. eg, _ := errgroup.WithContext(ctx)
  408. w := progress.ContextWriter(ctx)
  409. for dep, config := range dependencies {
  410. if shouldWait, err := shouldWaitForDependency(dep, config, project); err != nil {
  411. return err
  412. } else if !shouldWait {
  413. continue
  414. }
  415. waitingFor := containers.filter(isService(dep), isNotOneOff)
  416. w.Events(containerEvents(waitingFor, progress.Waiting))
  417. if len(waitingFor) == 0 {
  418. if config.Required {
  419. return fmt.Errorf("%s is missing dependency %s", dependant, dep)
  420. }
  421. logrus.Warnf("%s is missing dependency %s", dependant, dep)
  422. continue
  423. }
  424. eg.Go(func() error {
  425. ticker := time.NewTicker(500 * time.Millisecond)
  426. defer ticker.Stop()
  427. for {
  428. select {
  429. case <-ticker.C:
  430. case <-ctx.Done():
  431. return nil
  432. }
  433. switch config.Condition {
  434. case ServiceConditionRunningOrHealthy:
  435. healthy, err := s.isServiceHealthy(ctx, waitingFor, true)
  436. if err != nil {
  437. if !config.Required {
  438. w.Events(containerReasonEvents(waitingFor, progress.SkippedEvent, fmt.Sprintf("optional dependency %q is not running or is unhealthy", dep)))
  439. logrus.Warnf("optional dependency %q is not running or is unhealthy: %s", dep, err.Error())
  440. return nil
  441. }
  442. return err
  443. }
  444. if healthy {
  445. w.Events(containerEvents(waitingFor, progress.Healthy))
  446. return nil
  447. }
  448. case types.ServiceConditionHealthy:
  449. healthy, err := s.isServiceHealthy(ctx, waitingFor, false)
  450. if err != nil {
  451. if !config.Required {
  452. w.Events(containerReasonEvents(waitingFor, progress.SkippedEvent, fmt.Sprintf("optional dependency %q failed to start", dep)))
  453. logrus.Warnf("optional dependency %q failed to start: %s", dep, err.Error())
  454. return nil
  455. }
  456. w.Events(containerEvents(waitingFor, progress.ErrorEvent))
  457. return fmt.Errorf("dependency failed to start: %w", err)
  458. }
  459. if healthy {
  460. w.Events(containerEvents(waitingFor, progress.Healthy))
  461. return nil
  462. }
  463. case types.ServiceConditionCompletedSuccessfully:
  464. exited, code, err := s.isServiceCompleted(ctx, waitingFor)
  465. if err != nil {
  466. return err
  467. }
  468. if exited {
  469. if code == 0 {
  470. w.Events(containerEvents(waitingFor, progress.Exited))
  471. return nil
  472. }
  473. messageSuffix := fmt.Sprintf("%q didn't complete successfully: exit %d", dep, code)
  474. if !config.Required {
  475. // optional -> mark as skipped & don't propagate error
  476. w.Events(containerReasonEvents(waitingFor, progress.SkippedEvent, fmt.Sprintf("optional dependency %s", messageSuffix)))
  477. logrus.Warnf("optional dependency %s", messageSuffix)
  478. return nil
  479. }
  480. msg := fmt.Sprintf("service %s", messageSuffix)
  481. w.Events(containerReasonEvents(waitingFor, progress.ErrorMessageEvent, msg))
  482. return errors.New(msg)
  483. }
  484. default:
  485. logrus.Warnf("unsupported depends_on condition: %s", config.Condition)
  486. return nil
  487. }
  488. }
  489. })
  490. }
  491. err := eg.Wait()
  492. if errors.Is(err, context.DeadlineExceeded) {
  493. return fmt.Errorf("timeout waiting for dependencies")
  494. }
  495. return err
  496. }
  497. func shouldWaitForDependency(serviceName string, dependencyConfig types.ServiceDependency, project *types.Project) (bool, error) {
  498. if dependencyConfig.Condition == types.ServiceConditionStarted {
  499. // already managed by InDependencyOrder
  500. return false, nil
  501. }
  502. if service, err := project.GetService(serviceName); err != nil {
  503. for _, ds := range project.DisabledServices {
  504. if ds.Name == serviceName {
  505. // don't wait for disabled service (--no-deps)
  506. return false, nil
  507. }
  508. }
  509. return false, err
  510. } else if service.GetScale() == 0 {
  511. // don't wait for the dependency which configured to have 0 containers running
  512. return false, nil
  513. } else if service.Provider != nil {
  514. // don't wait for provider services
  515. return false, nil
  516. }
  517. return true, nil
  518. }
  519. func nextContainerNumber(containers []containerType.Summary) int {
  520. maxNumber := 0
  521. for _, c := range containers {
  522. s, ok := c.Labels[api.ContainerNumberLabel]
  523. if !ok {
  524. logrus.Warnf("container %s is missing %s label", c.ID, api.ContainerNumberLabel)
  525. }
  526. n, err := strconv.Atoi(s)
  527. if err != nil {
  528. logrus.Warnf("container %s has invalid %s label: %s", c.ID, api.ContainerNumberLabel, s)
  529. continue
  530. }
  531. if n > maxNumber {
  532. maxNumber = n
  533. }
  534. }
  535. return maxNumber + 1
  536. }
  537. func (s *composeService) createContainer(ctx context.Context, project *types.Project, service types.ServiceConfig,
  538. name string, number int, opts createOptions,
  539. ) (ctr containerType.Summary, err error) {
  540. w := progress.ContextWriter(ctx)
  541. eventName := "Container " + name
  542. w.Event(progress.CreatingEvent(eventName))
  543. ctr, err = s.createMobyContainer(ctx, project, service, name, number, nil, opts, w)
  544. if err != nil {
  545. if ctx.Err() == nil {
  546. w.Event(progress.Event{
  547. ID: eventName,
  548. Status: progress.Error,
  549. StatusText: err.Error(),
  550. })
  551. }
  552. return
  553. }
  554. w.Event(progress.CreatedEvent(eventName))
  555. return
  556. }
  557. func (s *composeService) recreateContainer(ctx context.Context, project *types.Project, service types.ServiceConfig,
  558. replaced containerType.Summary, inherit bool, timeout *time.Duration,
  559. ) (created containerType.Summary, err error) {
  560. w := progress.ContextWriter(ctx)
  561. eventName := getContainerProgressName(replaced)
  562. w.Event(progress.NewEvent(eventName, progress.Working, "Recreate"))
  563. defer func() {
  564. if err != nil && ctx.Err() == nil {
  565. w.Event(progress.Event{
  566. ID: eventName,
  567. Status: progress.Error,
  568. StatusText: err.Error(),
  569. })
  570. }
  571. }()
  572. number, err := strconv.Atoi(replaced.Labels[api.ContainerNumberLabel])
  573. if err != nil {
  574. return created, err
  575. }
  576. var inherited *containerType.Summary
  577. if inherit {
  578. inherited = &replaced
  579. }
  580. replacedContainerName := service.ContainerName
  581. if replacedContainerName == "" {
  582. replacedContainerName = service.Name + api.Separator + strconv.Itoa(number)
  583. }
  584. name := getContainerName(project.Name, service, number)
  585. tmpName := fmt.Sprintf("%s_%s", replaced.ID[:12], name)
  586. opts := createOptions{
  587. AutoRemove: false,
  588. AttachStdin: false,
  589. UseNetworkAliases: true,
  590. Labels: mergeLabels(service.Labels, service.CustomLabels).Add(api.ContainerReplaceLabel, replacedContainerName),
  591. }
  592. created, err = s.createMobyContainer(ctx, project, service, tmpName, number, inherited, opts, w)
  593. if err != nil {
  594. return created, err
  595. }
  596. timeoutInSecond := utils.DurationSecondToInt(timeout)
  597. err = s.apiClient().ContainerStop(ctx, replaced.ID, containerType.StopOptions{Timeout: timeoutInSecond})
  598. if err != nil {
  599. return created, err
  600. }
  601. err = s.apiClient().ContainerRemove(ctx, replaced.ID, containerType.RemoveOptions{})
  602. if err != nil {
  603. return created, err
  604. }
  605. err = s.apiClient().ContainerRename(ctx, tmpName, name)
  606. if err != nil {
  607. return created, err
  608. }
  609. w.Event(progress.NewEvent(eventName, progress.Done, "Recreated"))
  610. return created, err
  611. }
  612. // force sequential calls to ContainerStart to prevent race condition in engine assigning ports from ranges
  613. var startMx sync.Mutex
  614. func (s *composeService) startContainer(ctx context.Context, ctr containerType.Summary) error {
  615. w := progress.ContextWriter(ctx)
  616. w.Event(progress.NewEvent(getContainerProgressName(ctr), progress.Working, "Restart"))
  617. startMx.Lock()
  618. defer startMx.Unlock()
  619. err := s.apiClient().ContainerStart(ctx, ctr.ID, containerType.StartOptions{})
  620. if err != nil {
  621. return err
  622. }
  623. w.Event(progress.NewEvent(getContainerProgressName(ctr), progress.Done, "Restarted"))
  624. return nil
  625. }
  626. func (s *composeService) createMobyContainer(ctx context.Context,
  627. project *types.Project,
  628. service types.ServiceConfig,
  629. name string,
  630. number int,
  631. inherit *containerType.Summary,
  632. opts createOptions,
  633. w progress.Writer,
  634. ) (containerType.Summary, error) {
  635. var created containerType.Summary
  636. cfgs, err := s.getCreateConfigs(ctx, project, service, number, inherit, opts)
  637. if err != nil {
  638. return created, err
  639. }
  640. platform := service.Platform
  641. if platform == "" {
  642. platform = project.Environment["DOCKER_DEFAULT_PLATFORM"]
  643. }
  644. var plat *specs.Platform
  645. if platform != "" {
  646. var p specs.Platform
  647. p, err = platforms.Parse(platform)
  648. if err != nil {
  649. return created, err
  650. }
  651. plat = &p
  652. }
  653. response, err := s.apiClient().ContainerCreate(ctx, cfgs.Container, cfgs.Host, cfgs.Network, plat, name)
  654. if err != nil {
  655. return created, err
  656. }
  657. for _, warning := range response.Warnings {
  658. w.Event(progress.Event{
  659. ID: service.Name,
  660. Status: progress.Warning,
  661. Text: warning,
  662. })
  663. }
  664. inspectedContainer, err := s.apiClient().ContainerInspect(ctx, response.ID)
  665. if err != nil {
  666. return created, err
  667. }
  668. created = containerType.Summary{
  669. ID: inspectedContainer.ID,
  670. Labels: inspectedContainer.Config.Labels,
  671. Names: []string{inspectedContainer.Name},
  672. NetworkSettings: &containerType.NetworkSettingsSummary{
  673. Networks: inspectedContainer.NetworkSettings.Networks,
  674. },
  675. }
  676. apiVersion, err := s.RuntimeVersion(ctx)
  677. if err != nil {
  678. return created, err
  679. }
  680. // Starting API version 1.44, the ContainerCreate API call takes multiple networks
  681. // so we include all the configurations there and can skip the one-by-one calls here
  682. if versions.LessThan(apiVersion, "1.44") {
  683. // the highest-priority network is the primary and is included in the ContainerCreate API
  684. // call via container.NetworkMode & network.NetworkingConfig
  685. // any remaining networks are connected one-by-one here after creation (but before start)
  686. serviceNetworks := service.NetworksByPriority()
  687. for _, networkKey := range serviceNetworks {
  688. mobyNetworkName := project.Networks[networkKey].Name
  689. if string(cfgs.Host.NetworkMode) == mobyNetworkName {
  690. // primary network already configured as part of ContainerCreate
  691. continue
  692. }
  693. epSettings := createEndpointSettings(project, service, number, networkKey, cfgs.Links, opts.UseNetworkAliases)
  694. if err := s.apiClient().NetworkConnect(ctx, mobyNetworkName, created.ID, epSettings); err != nil {
  695. return created, err
  696. }
  697. }
  698. }
  699. return created, nil
  700. }
  701. // getLinks mimics V1 compose/service.py::Service::_get_links()
  702. func (s *composeService) getLinks(ctx context.Context, projectName string, service types.ServiceConfig, number int) ([]string, error) {
  703. var links []string
  704. format := func(k, v string) string {
  705. return fmt.Sprintf("%s:%s", k, v)
  706. }
  707. getServiceContainers := func(serviceName string) (Containers, error) {
  708. return s.getContainers(ctx, projectName, oneOffExclude, true, serviceName)
  709. }
  710. for _, rawLink := range service.Links {
  711. linkSplit := strings.Split(rawLink, ":")
  712. linkServiceName := linkSplit[0]
  713. linkName := linkServiceName
  714. if len(linkSplit) == 2 {
  715. linkName = linkSplit[1] // linkName if informed like in: "serviceName:linkName"
  716. }
  717. cnts, err := getServiceContainers(linkServiceName)
  718. if err != nil {
  719. return nil, err
  720. }
  721. for _, c := range cnts {
  722. containerName := getCanonicalContainerName(c)
  723. links = append(links,
  724. format(containerName, linkName),
  725. format(containerName, linkServiceName+api.Separator+strconv.Itoa(number)),
  726. format(containerName, strings.Join([]string{projectName, linkServiceName, strconv.Itoa(number)}, api.Separator)),
  727. )
  728. }
  729. }
  730. if service.Labels[api.OneoffLabel] == "True" {
  731. cnts, err := getServiceContainers(service.Name)
  732. if err != nil {
  733. return nil, err
  734. }
  735. for _, c := range cnts {
  736. containerName := getCanonicalContainerName(c)
  737. links = append(links,
  738. format(containerName, service.Name),
  739. format(containerName, strings.TrimPrefix(containerName, projectName+api.Separator)),
  740. format(containerName, containerName),
  741. )
  742. }
  743. }
  744. for _, rawExtLink := range service.ExternalLinks {
  745. extLinkSplit := strings.Split(rawExtLink, ":")
  746. externalLink := extLinkSplit[0]
  747. linkName := externalLink
  748. if len(extLinkSplit) == 2 {
  749. linkName = extLinkSplit[1]
  750. }
  751. links = append(links, format(externalLink, linkName))
  752. }
  753. return links, nil
  754. }
  755. func (s *composeService) isServiceHealthy(ctx context.Context, containers Containers, fallbackRunning bool) (bool, error) {
  756. for _, c := range containers {
  757. ctr, err := s.apiClient().ContainerInspect(ctx, c.ID)
  758. if err != nil {
  759. return false, err
  760. }
  761. name := ctr.Name[1:]
  762. if ctr.State.Status == containerType.StateExited {
  763. return false, fmt.Errorf("container %s exited (%d)", name, ctr.State.ExitCode)
  764. }
  765. if ctr.Config.Healthcheck == nil && fallbackRunning {
  766. // Container does not define a health check, but we can fall back to "running" state
  767. return ctr.State != nil && ctr.State.Status == containerType.StateRunning, nil
  768. }
  769. if ctr.State == nil || ctr.State.Health == nil {
  770. return false, fmt.Errorf("container %s has no healthcheck configured", name)
  771. }
  772. switch ctr.State.Health.Status {
  773. case containerType.Healthy:
  774. // Continue by checking the next container.
  775. case containerType.Unhealthy:
  776. return false, fmt.Errorf("container %s is unhealthy", name)
  777. case containerType.Starting:
  778. return false, nil
  779. default:
  780. return false, fmt.Errorf("container %s had unexpected health status %q", name, ctr.State.Health.Status)
  781. }
  782. }
  783. return true, nil
  784. }
  785. func (s *composeService) isServiceCompleted(ctx context.Context, containers Containers) (bool, int, error) {
  786. for _, c := range containers {
  787. ctr, err := s.apiClient().ContainerInspect(ctx, c.ID)
  788. if err != nil {
  789. return false, 0, err
  790. }
  791. if ctr.State != nil && ctr.State.Status == containerType.StateExited {
  792. return true, ctr.State.ExitCode, nil
  793. }
  794. }
  795. return false, 0, nil
  796. }
  797. func (s *composeService) startService(ctx context.Context,
  798. project *types.Project, service types.ServiceConfig,
  799. containers Containers, listener api.ContainerEventListener,
  800. timeout time.Duration,
  801. ) error {
  802. if service.Deploy != nil && service.Deploy.Replicas != nil && *service.Deploy.Replicas == 0 {
  803. return nil
  804. }
  805. err := s.waitDependencies(ctx, project, service.Name, service.DependsOn, containers, timeout)
  806. if err != nil {
  807. return err
  808. }
  809. if len(containers) == 0 {
  810. if service.GetScale() == 0 {
  811. return nil
  812. }
  813. return fmt.Errorf("service %q has no container to start", service.Name)
  814. }
  815. w := progress.ContextWriter(ctx)
  816. for _, ctr := range containers.filter(isService(service.Name)) {
  817. if ctr.State == ContainerRunning {
  818. continue
  819. }
  820. err = s.injectSecrets(ctx, project, service, ctr.ID)
  821. if err != nil {
  822. return err
  823. }
  824. err = s.injectConfigs(ctx, project, service, ctr.ID)
  825. if err != nil {
  826. return err
  827. }
  828. eventName := getContainerProgressName(ctr)
  829. w.Event(progress.StartingEvent(eventName))
  830. err = s.apiClient().ContainerStart(ctx, ctr.ID, containerType.StartOptions{})
  831. if err != nil {
  832. return err
  833. }
  834. for _, hook := range service.PostStart {
  835. err = s.runHook(ctx, ctr, service, hook, listener)
  836. if err != nil {
  837. return err
  838. }
  839. }
  840. w.Event(progress.StartedEvent(eventName))
  841. }
  842. return nil
  843. }
  844. func mergeLabels(ls ...types.Labels) types.Labels {
  845. merged := types.Labels{}
  846. for _, l := range ls {
  847. maps.Copy(merged, l)
  848. }
  849. return merged
  850. }