service.go 28 KB

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