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