convergence.go 25 KB

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