convergence.go 26 KB

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