convergence.go 25 KB

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