convergence.go 26 KB

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