service.go 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939
  1. package controlapi
  2. import (
  3. "errors"
  4. "reflect"
  5. "strings"
  6. "time"
  7. "github.com/docker/distribution/reference"
  8. "github.com/docker/swarmkit/api"
  9. "github.com/docker/swarmkit/api/defaults"
  10. "github.com/docker/swarmkit/api/genericresource"
  11. "github.com/docker/swarmkit/api/naming"
  12. "github.com/docker/swarmkit/identity"
  13. "github.com/docker/swarmkit/manager/allocator"
  14. "github.com/docker/swarmkit/manager/constraint"
  15. "github.com/docker/swarmkit/manager/state/store"
  16. "github.com/docker/swarmkit/protobuf/ptypes"
  17. "github.com/docker/swarmkit/template"
  18. gogotypes "github.com/gogo/protobuf/types"
  19. "golang.org/x/net/context"
  20. "google.golang.org/grpc"
  21. "google.golang.org/grpc/codes"
  22. )
  23. var (
  24. errNetworkUpdateNotSupported = errors.New("networks must be migrated to TaskSpec before being changed")
  25. errRenameNotSupported = errors.New("renaming services is not supported")
  26. errModeChangeNotAllowed = errors.New("service mode change is not allowed")
  27. )
  28. const minimumDuration = 1 * time.Millisecond
  29. func validateResources(r *api.Resources) error {
  30. if r == nil {
  31. return nil
  32. }
  33. if r.NanoCPUs != 0 && r.NanoCPUs < 1e6 {
  34. return grpc.Errorf(codes.InvalidArgument, "invalid cpu value %g: Must be at least %g", float64(r.NanoCPUs)/1e9, 1e6/1e9)
  35. }
  36. if r.MemoryBytes != 0 && r.MemoryBytes < 4*1024*1024 {
  37. return grpc.Errorf(codes.InvalidArgument, "invalid memory value %d: Must be at least 4MiB", r.MemoryBytes)
  38. }
  39. if err := genericresource.ValidateTask(r); err != nil {
  40. return nil
  41. }
  42. return nil
  43. }
  44. func validateResourceRequirements(r *api.ResourceRequirements) error {
  45. if r == nil {
  46. return nil
  47. }
  48. if err := validateResources(r.Limits); err != nil {
  49. return err
  50. }
  51. if err := validateResources(r.Reservations); err != nil {
  52. return err
  53. }
  54. return nil
  55. }
  56. func validateRestartPolicy(rp *api.RestartPolicy) error {
  57. if rp == nil {
  58. return nil
  59. }
  60. if rp.Delay != nil {
  61. delay, err := gogotypes.DurationFromProto(rp.Delay)
  62. if err != nil {
  63. return err
  64. }
  65. if delay < 0 {
  66. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: restart-delay cannot be negative")
  67. }
  68. }
  69. if rp.Window != nil {
  70. win, err := gogotypes.DurationFromProto(rp.Window)
  71. if err != nil {
  72. return err
  73. }
  74. if win < 0 {
  75. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: restart-window cannot be negative")
  76. }
  77. }
  78. return nil
  79. }
  80. func validatePlacement(placement *api.Placement) error {
  81. if placement == nil {
  82. return nil
  83. }
  84. _, err := constraint.Parse(placement.Constraints)
  85. return err
  86. }
  87. func validateUpdate(uc *api.UpdateConfig) error {
  88. if uc == nil {
  89. return nil
  90. }
  91. if uc.Delay < 0 {
  92. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: update-delay cannot be negative")
  93. }
  94. if uc.Monitor != nil {
  95. monitor, err := gogotypes.DurationFromProto(uc.Monitor)
  96. if err != nil {
  97. return err
  98. }
  99. if monitor < 0 {
  100. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: update-monitor cannot be negative")
  101. }
  102. }
  103. if uc.MaxFailureRatio < 0 || uc.MaxFailureRatio > 1 {
  104. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: update-maxfailureratio cannot be less than 0 or bigger than 1")
  105. }
  106. return nil
  107. }
  108. func validateContainerSpec(taskSpec api.TaskSpec) error {
  109. // Building a empty/dummy Task to validate the templating and
  110. // the resulting container spec as well. This is a *best effort*
  111. // validation.
  112. container, err := template.ExpandContainerSpec(&api.NodeDescription{
  113. Hostname: "nodeHostname",
  114. Platform: &api.Platform{
  115. OS: "os",
  116. Architecture: "architecture",
  117. },
  118. }, &api.Task{
  119. Spec: taskSpec,
  120. ServiceID: "serviceid",
  121. Slot: 1,
  122. NodeID: "nodeid",
  123. Networks: []*api.NetworkAttachment{},
  124. Annotations: api.Annotations{
  125. Name: "taskname",
  126. },
  127. ServiceAnnotations: api.Annotations{
  128. Name: "servicename",
  129. },
  130. Endpoint: &api.Endpoint{},
  131. LogDriver: taskSpec.LogDriver,
  132. })
  133. if err != nil {
  134. return grpc.Errorf(codes.InvalidArgument, err.Error())
  135. }
  136. if err := validateImage(container.Image); err != nil {
  137. return err
  138. }
  139. if err := validateMounts(container.Mounts); err != nil {
  140. return err
  141. }
  142. if err := validateHealthCheck(container.Healthcheck); err != nil {
  143. return err
  144. }
  145. return nil
  146. }
  147. // validateImage validates image name in containerSpec
  148. func validateImage(image string) error {
  149. if image == "" {
  150. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: image reference must be provided")
  151. }
  152. if _, err := reference.ParseNormalizedNamed(image); err != nil {
  153. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: %q is not a valid repository/tag", image)
  154. }
  155. return nil
  156. }
  157. // validateMounts validates if there are duplicate mounts in containerSpec
  158. func validateMounts(mounts []api.Mount) error {
  159. mountMap := make(map[string]bool)
  160. for _, mount := range mounts {
  161. if _, exists := mountMap[mount.Target]; exists {
  162. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: duplicate mount point: %s", mount.Target)
  163. }
  164. mountMap[mount.Target] = true
  165. }
  166. return nil
  167. }
  168. // validateHealthCheck validates configs about container's health check
  169. func validateHealthCheck(hc *api.HealthConfig) error {
  170. if hc == nil {
  171. return nil
  172. }
  173. if hc.Interval != nil {
  174. interval, err := gogotypes.DurationFromProto(hc.Interval)
  175. if err != nil {
  176. return err
  177. }
  178. if interval != 0 && interval < time.Duration(minimumDuration) {
  179. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: Interval in HealthConfig cannot be less than %s", minimumDuration)
  180. }
  181. }
  182. if hc.Timeout != nil {
  183. timeout, err := gogotypes.DurationFromProto(hc.Timeout)
  184. if err != nil {
  185. return err
  186. }
  187. if timeout != 0 && timeout < time.Duration(minimumDuration) {
  188. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: Timeout in HealthConfig cannot be less than %s", minimumDuration)
  189. }
  190. }
  191. if hc.StartPeriod != nil {
  192. sp, err := gogotypes.DurationFromProto(hc.StartPeriod)
  193. if err != nil {
  194. return err
  195. }
  196. if sp != 0 && sp < time.Duration(minimumDuration) {
  197. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: StartPeriod in HealthConfig cannot be less than %s", minimumDuration)
  198. }
  199. }
  200. if hc.Retries < 0 {
  201. return grpc.Errorf(codes.InvalidArgument, "ContainerSpec: Retries in HealthConfig cannot be negative")
  202. }
  203. return nil
  204. }
  205. func validateGenericRuntimeSpec(taskSpec api.TaskSpec) error {
  206. generic := taskSpec.GetGeneric()
  207. if len(generic.Kind) < 3 {
  208. return grpc.Errorf(codes.InvalidArgument, "Generic runtime: Invalid name %q", generic.Kind)
  209. }
  210. reservedNames := []string{"container", "attachment"}
  211. for _, n := range reservedNames {
  212. if strings.ToLower(generic.Kind) == n {
  213. return grpc.Errorf(codes.InvalidArgument, "Generic runtime: %q is a reserved name", generic.Kind)
  214. }
  215. }
  216. payload := generic.Payload
  217. if payload == nil {
  218. return grpc.Errorf(codes.InvalidArgument, "Generic runtime is missing payload")
  219. }
  220. if payload.TypeUrl == "" {
  221. return grpc.Errorf(codes.InvalidArgument, "Generic runtime is missing payload type")
  222. }
  223. if len(payload.Value) == 0 {
  224. return grpc.Errorf(codes.InvalidArgument, "Generic runtime has an empty payload")
  225. }
  226. return nil
  227. }
  228. func validateTaskSpec(taskSpec api.TaskSpec) error {
  229. if err := validateResourceRequirements(taskSpec.Resources); err != nil {
  230. return err
  231. }
  232. if err := validateRestartPolicy(taskSpec.Restart); err != nil {
  233. return err
  234. }
  235. if err := validatePlacement(taskSpec.Placement); err != nil {
  236. return err
  237. }
  238. // Check to see if the secret reference portion of the spec is valid
  239. if err := validateSecretRefsSpec(taskSpec); err != nil {
  240. return err
  241. }
  242. // Check to see if the config reference portion of the spec is valid
  243. if err := validateConfigRefsSpec(taskSpec); err != nil {
  244. return err
  245. }
  246. if taskSpec.GetRuntime() == nil {
  247. return grpc.Errorf(codes.InvalidArgument, "TaskSpec: missing runtime")
  248. }
  249. switch taskSpec.GetRuntime().(type) {
  250. case *api.TaskSpec_Container:
  251. if err := validateContainerSpec(taskSpec); err != nil {
  252. return err
  253. }
  254. case *api.TaskSpec_Generic:
  255. if err := validateGenericRuntimeSpec(taskSpec); err != nil {
  256. return err
  257. }
  258. default:
  259. return grpc.Errorf(codes.Unimplemented, "RuntimeSpec: unimplemented runtime in service spec")
  260. }
  261. return nil
  262. }
  263. func validateEndpointSpec(epSpec *api.EndpointSpec) error {
  264. // Endpoint spec is optional
  265. if epSpec == nil {
  266. return nil
  267. }
  268. type portSpec struct {
  269. publishedPort uint32
  270. protocol api.PortConfig_Protocol
  271. }
  272. portSet := make(map[portSpec]struct{})
  273. for _, port := range epSpec.Ports {
  274. // Publish mode = "ingress" represents Routing-Mesh and current implementation
  275. // of routing-mesh relies on IPVS based load-balancing with input=published-port.
  276. // But Endpoint-Spec mode of DNSRR relies on multiple A records and cannot be used
  277. // with routing-mesh (PublishMode="ingress") which cannot rely on DNSRR.
  278. // But PublishMode="host" doesn't provide Routing-Mesh and the DNSRR is applicable
  279. // for the backend network and hence we accept that configuration.
  280. if epSpec.Mode == api.ResolutionModeDNSRoundRobin && port.PublishMode == api.PublishModeIngress {
  281. return grpc.Errorf(codes.InvalidArgument, "EndpointSpec: port published with ingress mode can't be used with dnsrr mode")
  282. }
  283. // If published port is not specified, it does not conflict
  284. // with any others.
  285. if port.PublishedPort == 0 {
  286. continue
  287. }
  288. portSpec := portSpec{publishedPort: port.PublishedPort, protocol: port.Protocol}
  289. if _, ok := portSet[portSpec]; ok {
  290. return grpc.Errorf(codes.InvalidArgument, "EndpointSpec: duplicate published ports provided")
  291. }
  292. portSet[portSpec] = struct{}{}
  293. }
  294. return nil
  295. }
  296. // validateSecretRefsSpec finds if the secrets passed in spec are valid and have no
  297. // conflicting targets.
  298. func validateSecretRefsSpec(spec api.TaskSpec) error {
  299. container := spec.GetContainer()
  300. if container == nil {
  301. return nil
  302. }
  303. // Keep a map to track all the targets that will be exposed
  304. // The string returned is only used for logging. It could as well be struct{}{}
  305. existingTargets := make(map[string]string)
  306. for _, secretRef := range container.Secrets {
  307. // SecretID and SecretName are mandatory, we have invalid references without them
  308. if secretRef.SecretID == "" || secretRef.SecretName == "" {
  309. return grpc.Errorf(codes.InvalidArgument, "malformed secret reference")
  310. }
  311. // Every secret reference requires a Target
  312. if secretRef.GetTarget() == nil {
  313. return grpc.Errorf(codes.InvalidArgument, "malformed secret reference, no target provided")
  314. }
  315. // If this is a file target, we will ensure filename uniqueness
  316. if secretRef.GetFile() != nil {
  317. fileName := secretRef.GetFile().Name
  318. if fileName == "" {
  319. return grpc.Errorf(codes.InvalidArgument, "malformed file secret reference, invalid target file name provided")
  320. }
  321. // If this target is already in use, we have conflicting targets
  322. if prevSecretName, ok := existingTargets[fileName]; ok {
  323. return grpc.Errorf(codes.InvalidArgument, "secret references '%s' and '%s' have a conflicting target: '%s'", prevSecretName, secretRef.SecretName, fileName)
  324. }
  325. existingTargets[fileName] = secretRef.SecretName
  326. }
  327. }
  328. return nil
  329. }
  330. // validateConfigRefsSpec finds if the configs passed in spec are valid and have no
  331. // conflicting targets.
  332. func validateConfigRefsSpec(spec api.TaskSpec) error {
  333. container := spec.GetContainer()
  334. if container == nil {
  335. return nil
  336. }
  337. // Keep a map to track all the targets that will be exposed
  338. // The string returned is only used for logging. It could as well be struct{}{}
  339. existingTargets := make(map[string]string)
  340. for _, configRef := range container.Configs {
  341. // ConfigID and ConfigName are mandatory, we have invalid references without them
  342. if configRef.ConfigID == "" || configRef.ConfigName == "" {
  343. return grpc.Errorf(codes.InvalidArgument, "malformed config reference")
  344. }
  345. // Every config reference requires a Target
  346. if configRef.GetTarget() == nil {
  347. return grpc.Errorf(codes.InvalidArgument, "malformed config reference, no target provided")
  348. }
  349. // If this is a file target, we will ensure filename uniqueness
  350. if configRef.GetFile() != nil {
  351. fileName := configRef.GetFile().Name
  352. // Validate the file name
  353. if fileName == "" {
  354. return grpc.Errorf(codes.InvalidArgument, "malformed file config reference, invalid target file name provided")
  355. }
  356. // If this target is already in use, we have conflicting targets
  357. if prevConfigName, ok := existingTargets[fileName]; ok {
  358. return grpc.Errorf(codes.InvalidArgument, "config references '%s' and '%s' have a conflicting target: '%s'", prevConfigName, configRef.ConfigName, fileName)
  359. }
  360. existingTargets[fileName] = configRef.ConfigName
  361. }
  362. }
  363. return nil
  364. }
  365. func (s *Server) validateNetworks(networks []*api.NetworkAttachmentConfig) error {
  366. for _, na := range networks {
  367. var network *api.Network
  368. s.store.View(func(tx store.ReadTx) {
  369. network = store.GetNetwork(tx, na.Target)
  370. })
  371. if network == nil {
  372. continue
  373. }
  374. if allocator.IsIngressNetwork(network) {
  375. return grpc.Errorf(codes.InvalidArgument,
  376. "Service cannot be explicitly attached to the ingress network %q", network.Spec.Annotations.Name)
  377. }
  378. }
  379. return nil
  380. }
  381. func validateMode(s *api.ServiceSpec) error {
  382. m := s.GetMode()
  383. switch m.(type) {
  384. case *api.ServiceSpec_Replicated:
  385. if int64(m.(*api.ServiceSpec_Replicated).Replicated.Replicas) < 0 {
  386. return grpc.Errorf(codes.InvalidArgument, "Number of replicas must be non-negative")
  387. }
  388. case *api.ServiceSpec_Global:
  389. default:
  390. return grpc.Errorf(codes.InvalidArgument, "Unrecognized service mode")
  391. }
  392. return nil
  393. }
  394. func validateServiceSpec(spec *api.ServiceSpec) error {
  395. if spec == nil {
  396. return grpc.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
  397. }
  398. if err := validateAnnotations(spec.Annotations); err != nil {
  399. return err
  400. }
  401. if err := validateTaskSpec(spec.Task); err != nil {
  402. return err
  403. }
  404. if err := validateUpdate(spec.Update); err != nil {
  405. return err
  406. }
  407. if err := validateEndpointSpec(spec.Endpoint); err != nil {
  408. return err
  409. }
  410. if err := validateMode(spec); err != nil {
  411. return err
  412. }
  413. return nil
  414. }
  415. // checkPortConflicts does a best effort to find if the passed in spec has port
  416. // conflicts with existing services.
  417. // `serviceID string` is the service ID of the spec in service update. If
  418. // `serviceID` is not "", then conflicts check will be skipped against this
  419. // service (the service being updated).
  420. func (s *Server) checkPortConflicts(spec *api.ServiceSpec, serviceID string) error {
  421. if spec.Endpoint == nil {
  422. return nil
  423. }
  424. type portSpec struct {
  425. protocol api.PortConfig_Protocol
  426. publishedPort uint32
  427. }
  428. pcToStruct := func(pc *api.PortConfig) portSpec {
  429. return portSpec{
  430. protocol: pc.Protocol,
  431. publishedPort: pc.PublishedPort,
  432. }
  433. }
  434. ingressPorts := make(map[portSpec]struct{})
  435. hostModePorts := make(map[portSpec]struct{})
  436. for _, pc := range spec.Endpoint.Ports {
  437. if pc.PublishedPort == 0 {
  438. continue
  439. }
  440. switch pc.PublishMode {
  441. case api.PublishModeIngress:
  442. ingressPorts[pcToStruct(pc)] = struct{}{}
  443. case api.PublishModeHost:
  444. hostModePorts[pcToStruct(pc)] = struct{}{}
  445. }
  446. }
  447. if len(ingressPorts) == 0 && len(hostModePorts) == 0 {
  448. return nil
  449. }
  450. var (
  451. services []*api.Service
  452. err error
  453. )
  454. s.store.View(func(tx store.ReadTx) {
  455. services, err = store.FindServices(tx, store.All)
  456. })
  457. if err != nil {
  458. return err
  459. }
  460. isPortInUse := func(pc *api.PortConfig, service *api.Service) error {
  461. if pc.PublishedPort == 0 {
  462. return nil
  463. }
  464. switch pc.PublishMode {
  465. case api.PublishModeHost:
  466. if _, ok := ingressPorts[pcToStruct(pc)]; ok {
  467. return grpc.Errorf(codes.InvalidArgument, "port '%d' is already in use by service '%s' (%s) as a host-published port", pc.PublishedPort, service.Spec.Annotations.Name, service.ID)
  468. }
  469. // Multiple services with same port in host publish mode can
  470. // coexist - this is handled by the scheduler.
  471. return nil
  472. case api.PublishModeIngress:
  473. _, ingressConflict := ingressPorts[pcToStruct(pc)]
  474. _, hostModeConflict := hostModePorts[pcToStruct(pc)]
  475. if ingressConflict || hostModeConflict {
  476. return grpc.Errorf(codes.InvalidArgument, "port '%d' is already in use by service '%s' (%s) as an ingress port", pc.PublishedPort, service.Spec.Annotations.Name, service.ID)
  477. }
  478. }
  479. return nil
  480. }
  481. for _, service := range services {
  482. // If service ID is the same (and not "") then this is an update
  483. if serviceID != "" && serviceID == service.ID {
  484. continue
  485. }
  486. if service.Spec.Endpoint != nil {
  487. for _, pc := range service.Spec.Endpoint.Ports {
  488. if err := isPortInUse(pc, service); err != nil {
  489. return err
  490. }
  491. }
  492. }
  493. if service.Endpoint != nil {
  494. for _, pc := range service.Endpoint.Ports {
  495. if err := isPortInUse(pc, service); err != nil {
  496. return err
  497. }
  498. }
  499. }
  500. }
  501. return nil
  502. }
  503. // checkSecretExistence finds if the secret exists
  504. func (s *Server) checkSecretExistence(tx store.Tx, spec *api.ServiceSpec) error {
  505. container := spec.Task.GetContainer()
  506. if container == nil {
  507. return nil
  508. }
  509. var failedSecrets []string
  510. for _, secretRef := range container.Secrets {
  511. secret := store.GetSecret(tx, secretRef.SecretID)
  512. // Check to see if the secret exists and secretRef.SecretName matches the actual secretName
  513. if secret == nil || secret.Spec.Annotations.Name != secretRef.SecretName {
  514. failedSecrets = append(failedSecrets, secretRef.SecretName)
  515. }
  516. }
  517. if len(failedSecrets) > 0 {
  518. secretStr := "secrets"
  519. if len(failedSecrets) == 1 {
  520. secretStr = "secret"
  521. }
  522. return grpc.Errorf(codes.InvalidArgument, "%s not found: %v", secretStr, strings.Join(failedSecrets, ", "))
  523. }
  524. return nil
  525. }
  526. // checkConfigExistence finds if the config exists
  527. func (s *Server) checkConfigExistence(tx store.Tx, spec *api.ServiceSpec) error {
  528. container := spec.Task.GetContainer()
  529. if container == nil {
  530. return nil
  531. }
  532. var failedConfigs []string
  533. for _, configRef := range container.Configs {
  534. config := store.GetConfig(tx, configRef.ConfigID)
  535. // Check to see if the config exists and configRef.ConfigName matches the actual configName
  536. if config == nil || config.Spec.Annotations.Name != configRef.ConfigName {
  537. failedConfigs = append(failedConfigs, configRef.ConfigName)
  538. }
  539. }
  540. if len(failedConfigs) > 0 {
  541. configStr := "configs"
  542. if len(failedConfigs) == 1 {
  543. configStr = "config"
  544. }
  545. return grpc.Errorf(codes.InvalidArgument, "%s not found: %v", configStr, strings.Join(failedConfigs, ", "))
  546. }
  547. return nil
  548. }
  549. // CreateService creates and returns a Service based on the provided ServiceSpec.
  550. // - Returns `InvalidArgument` if the ServiceSpec is malformed.
  551. // - Returns `Unimplemented` if the ServiceSpec references unimplemented features.
  552. // - Returns `AlreadyExists` if the ServiceID conflicts.
  553. // - Returns an error if the creation fails.
  554. func (s *Server) CreateService(ctx context.Context, request *api.CreateServiceRequest) (*api.CreateServiceResponse, error) {
  555. if err := validateServiceSpec(request.Spec); err != nil {
  556. return nil, err
  557. }
  558. if err := s.validateNetworks(request.Spec.Networks); err != nil {
  559. return nil, err
  560. }
  561. if err := s.checkPortConflicts(request.Spec, ""); err != nil {
  562. return nil, err
  563. }
  564. // TODO(aluzzardi): Consider using `Name` as a primary key to handle
  565. // duplicate creations. See #65
  566. service := &api.Service{
  567. ID: identity.NewID(),
  568. Spec: *request.Spec,
  569. SpecVersion: &api.Version{},
  570. }
  571. if allocator.IsIngressNetworkNeeded(service) {
  572. if _, err := allocator.GetIngressNetwork(s.store); err == allocator.ErrNoIngress {
  573. return nil, grpc.Errorf(codes.FailedPrecondition, "service needs ingress network, but no ingress network is present")
  574. }
  575. }
  576. err := s.store.Update(func(tx store.Tx) error {
  577. // Check to see if all the secrets being added exist as objects
  578. // in our datastore
  579. err := s.checkSecretExistence(tx, request.Spec)
  580. if err != nil {
  581. return err
  582. }
  583. err = s.checkConfigExistence(tx, request.Spec)
  584. if err != nil {
  585. return err
  586. }
  587. return store.CreateService(tx, service)
  588. })
  589. if err != nil {
  590. return nil, err
  591. }
  592. return &api.CreateServiceResponse{
  593. Service: service,
  594. }, nil
  595. }
  596. // GetService returns a Service given a ServiceID.
  597. // - Returns `InvalidArgument` if ServiceID is not provided.
  598. // - Returns `NotFound` if the Service is not found.
  599. func (s *Server) GetService(ctx context.Context, request *api.GetServiceRequest) (*api.GetServiceResponse, error) {
  600. if request.ServiceID == "" {
  601. return nil, grpc.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
  602. }
  603. var service *api.Service
  604. s.store.View(func(tx store.ReadTx) {
  605. service = store.GetService(tx, request.ServiceID)
  606. })
  607. if service == nil {
  608. return nil, grpc.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
  609. }
  610. if request.InsertDefaults {
  611. service.Spec = *defaults.InterpolateService(&service.Spec)
  612. }
  613. return &api.GetServiceResponse{
  614. Service: service,
  615. }, nil
  616. }
  617. // UpdateService updates a Service referenced by ServiceID with the given ServiceSpec.
  618. // - Returns `NotFound` if the Service is not found.
  619. // - Returns `InvalidArgument` if the ServiceSpec is malformed.
  620. // - Returns `Unimplemented` if the ServiceSpec references unimplemented features.
  621. // - Returns an error if the update fails.
  622. func (s *Server) UpdateService(ctx context.Context, request *api.UpdateServiceRequest) (*api.UpdateServiceResponse, error) {
  623. if request.ServiceID == "" || request.ServiceVersion == nil {
  624. return nil, grpc.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
  625. }
  626. if err := validateServiceSpec(request.Spec); err != nil {
  627. return nil, err
  628. }
  629. var service *api.Service
  630. s.store.View(func(tx store.ReadTx) {
  631. service = store.GetService(tx, request.ServiceID)
  632. })
  633. if service == nil {
  634. return nil, grpc.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
  635. }
  636. if request.Spec.Endpoint != nil && !reflect.DeepEqual(request.Spec.Endpoint, service.Spec.Endpoint) {
  637. if err := s.checkPortConflicts(request.Spec, request.ServiceID); err != nil {
  638. return nil, err
  639. }
  640. }
  641. err := s.store.Update(func(tx store.Tx) error {
  642. service = store.GetService(tx, request.ServiceID)
  643. if service == nil {
  644. return grpc.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
  645. }
  646. // It's not okay to update Service.Spec.Networks on its own.
  647. // However, if Service.Spec.Task.Networks is also being
  648. // updated, that's okay (for example when migrating from the
  649. // deprecated Spec.Networks field to Spec.Task.Networks).
  650. if (len(request.Spec.Networks) != 0 || len(service.Spec.Networks) != 0) &&
  651. !reflect.DeepEqual(request.Spec.Networks, service.Spec.Networks) &&
  652. reflect.DeepEqual(request.Spec.Task.Networks, service.Spec.Task.Networks) {
  653. return grpc.Errorf(codes.Unimplemented, errNetworkUpdateNotSupported.Error())
  654. }
  655. // Check to see if all the secrets being added exist as objects
  656. // in our datastore
  657. err := s.checkSecretExistence(tx, request.Spec)
  658. if err != nil {
  659. return err
  660. }
  661. err = s.checkConfigExistence(tx, request.Spec)
  662. if err != nil {
  663. return err
  664. }
  665. // orchestrator is designed to be stateless, so it should not deal
  666. // with service mode change (comparing current config with previous config).
  667. // proper way to change service mode is to delete and re-add.
  668. if reflect.TypeOf(service.Spec.Mode) != reflect.TypeOf(request.Spec.Mode) {
  669. return grpc.Errorf(codes.Unimplemented, errModeChangeNotAllowed.Error())
  670. }
  671. if service.Spec.Annotations.Name != request.Spec.Annotations.Name {
  672. return grpc.Errorf(codes.Unimplemented, errRenameNotSupported.Error())
  673. }
  674. service.Meta.Version = *request.ServiceVersion
  675. if request.Rollback == api.UpdateServiceRequest_PREVIOUS {
  676. if service.PreviousSpec == nil {
  677. return grpc.Errorf(codes.FailedPrecondition, "service %s does not have a previous spec", request.ServiceID)
  678. }
  679. curSpec := service.Spec.Copy()
  680. curSpecVersion := service.SpecVersion
  681. service.Spec = *service.PreviousSpec.Copy()
  682. service.SpecVersion = service.PreviousSpecVersion.Copy()
  683. service.PreviousSpec = curSpec
  684. service.PreviousSpecVersion = curSpecVersion
  685. service.UpdateStatus = &api.UpdateStatus{
  686. State: api.UpdateStatus_ROLLBACK_STARTED,
  687. Message: "manually requested rollback",
  688. StartedAt: ptypes.MustTimestampProto(time.Now()),
  689. }
  690. } else {
  691. service.PreviousSpec = service.Spec.Copy()
  692. service.PreviousSpecVersion = service.SpecVersion
  693. service.Spec = *request.Spec.Copy()
  694. // Set spec version. Note that this will not match the
  695. // service's Meta.Version after the store update. The
  696. // versions for the spec and the service itself are not
  697. // meant to be directly comparable.
  698. service.SpecVersion = service.Meta.Version.Copy()
  699. // Reset update status
  700. service.UpdateStatus = nil
  701. }
  702. if allocator.IsIngressNetworkNeeded(service) {
  703. if _, err := allocator.GetIngressNetwork(s.store); err == allocator.ErrNoIngress {
  704. return grpc.Errorf(codes.FailedPrecondition, "service needs ingress network, but no ingress network is present")
  705. }
  706. }
  707. return store.UpdateService(tx, service)
  708. })
  709. if err != nil {
  710. return nil, err
  711. }
  712. return &api.UpdateServiceResponse{
  713. Service: service,
  714. }, nil
  715. }
  716. // RemoveService removes a Service referenced by ServiceID.
  717. // - Returns `InvalidArgument` if ServiceID is not provided.
  718. // - Returns `NotFound` if the Service is not found.
  719. // - Returns an error if the deletion fails.
  720. func (s *Server) RemoveService(ctx context.Context, request *api.RemoveServiceRequest) (*api.RemoveServiceResponse, error) {
  721. if request.ServiceID == "" {
  722. return nil, grpc.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
  723. }
  724. err := s.store.Update(func(tx store.Tx) error {
  725. return store.DeleteService(tx, request.ServiceID)
  726. })
  727. if err != nil {
  728. if err == store.ErrNotExist {
  729. return nil, grpc.Errorf(codes.NotFound, "service %s not found", request.ServiceID)
  730. }
  731. return nil, err
  732. }
  733. return &api.RemoveServiceResponse{}, nil
  734. }
  735. func filterServices(candidates []*api.Service, filters ...func(*api.Service) bool) []*api.Service {
  736. result := []*api.Service{}
  737. for _, c := range candidates {
  738. match := true
  739. for _, f := range filters {
  740. if !f(c) {
  741. match = false
  742. break
  743. }
  744. }
  745. if match {
  746. result = append(result, c)
  747. }
  748. }
  749. return result
  750. }
  751. // ListServices returns a list of all services.
  752. func (s *Server) ListServices(ctx context.Context, request *api.ListServicesRequest) (*api.ListServicesResponse, error) {
  753. var (
  754. services []*api.Service
  755. err error
  756. )
  757. s.store.View(func(tx store.ReadTx) {
  758. switch {
  759. case request.Filters != nil && len(request.Filters.Names) > 0:
  760. services, err = store.FindServices(tx, buildFilters(store.ByName, request.Filters.Names))
  761. case request.Filters != nil && len(request.Filters.NamePrefixes) > 0:
  762. services, err = store.FindServices(tx, buildFilters(store.ByNamePrefix, request.Filters.NamePrefixes))
  763. case request.Filters != nil && len(request.Filters.IDPrefixes) > 0:
  764. services, err = store.FindServices(tx, buildFilters(store.ByIDPrefix, request.Filters.IDPrefixes))
  765. case request.Filters != nil && len(request.Filters.Runtimes) > 0:
  766. services, err = store.FindServices(tx, buildFilters(store.ByRuntime, request.Filters.Runtimes))
  767. default:
  768. services, err = store.FindServices(tx, store.All)
  769. }
  770. })
  771. if err != nil {
  772. return nil, err
  773. }
  774. if request.Filters != nil {
  775. services = filterServices(services,
  776. func(e *api.Service) bool {
  777. return filterContains(e.Spec.Annotations.Name, request.Filters.Names)
  778. },
  779. func(e *api.Service) bool {
  780. return filterContainsPrefix(e.Spec.Annotations.Name, request.Filters.NamePrefixes)
  781. },
  782. func(e *api.Service) bool {
  783. return filterContainsPrefix(e.ID, request.Filters.IDPrefixes)
  784. },
  785. func(e *api.Service) bool {
  786. return filterMatchLabels(e.Spec.Annotations.Labels, request.Filters.Labels)
  787. },
  788. func(e *api.Service) bool {
  789. if len(request.Filters.Runtimes) == 0 {
  790. return true
  791. }
  792. r, err := naming.Runtime(e.Spec.Task)
  793. if err != nil {
  794. return false
  795. }
  796. return filterContains(r, request.Filters.Runtimes)
  797. },
  798. )
  799. }
  800. return &api.ListServicesResponse{
  801. Services: services,
  802. }, nil
  803. }