convergence.go 28 KB

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