convergence.go 28 KB

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