convergence.go 27 KB

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