convergence.go 28 KB

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