swarm.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549
  1. package cluster
  2. import (
  3. "fmt"
  4. "net"
  5. "strings"
  6. "time"
  7. "github.com/Sirupsen/logrus"
  8. apierrors "github.com/docker/docker/api/errors"
  9. apitypes "github.com/docker/docker/api/types"
  10. "github.com/docker/docker/api/types/filters"
  11. types "github.com/docker/docker/api/types/swarm"
  12. "github.com/docker/docker/daemon/cluster/convert"
  13. "github.com/docker/docker/opts"
  14. "github.com/docker/docker/pkg/signal"
  15. swarmapi "github.com/docker/swarmkit/api"
  16. "github.com/docker/swarmkit/manager/encryption"
  17. swarmnode "github.com/docker/swarmkit/node"
  18. "github.com/pkg/errors"
  19. "golang.org/x/net/context"
  20. )
  21. // Init initializes new cluster from user provided request.
  22. func (c *Cluster) Init(req types.InitRequest) (string, error) {
  23. c.controlMutex.Lock()
  24. defer c.controlMutex.Unlock()
  25. if c.nr != nil {
  26. if req.ForceNewCluster {
  27. // Take c.mu temporarily to wait for presently running
  28. // API handlers to finish before shutting down the node.
  29. c.mu.Lock()
  30. c.mu.Unlock()
  31. if err := c.nr.Stop(); err != nil {
  32. return "", err
  33. }
  34. } else {
  35. return "", errSwarmExists
  36. }
  37. }
  38. if err := validateAndSanitizeInitRequest(&req); err != nil {
  39. return "", apierrors.NewBadRequestError(err)
  40. }
  41. listenHost, listenPort, err := resolveListenAddr(req.ListenAddr)
  42. if err != nil {
  43. return "", err
  44. }
  45. advertiseHost, advertisePort, err := c.resolveAdvertiseAddr(req.AdvertiseAddr, listenPort)
  46. if err != nil {
  47. return "", err
  48. }
  49. dataPathAddr, err := resolveDataPathAddr(req.DataPathAddr)
  50. if err != nil {
  51. return "", err
  52. }
  53. localAddr := listenHost
  54. // If the local address is undetermined, the advertise address
  55. // will be used as local address, if it belongs to this system.
  56. // If the advertise address is not local, then we try to find
  57. // a system address to use as local address. If this fails,
  58. // we give up and ask the user to pass the listen address.
  59. if net.ParseIP(localAddr).IsUnspecified() {
  60. advertiseIP := net.ParseIP(advertiseHost)
  61. found := false
  62. for _, systemIP := range listSystemIPs() {
  63. if systemIP.Equal(advertiseIP) {
  64. localAddr = advertiseIP.String()
  65. found = true
  66. break
  67. }
  68. }
  69. if !found {
  70. ip, err := c.resolveSystemAddr()
  71. if err != nil {
  72. logrus.Warnf("Could not find a local address: %v", err)
  73. return "", errMustSpecifyListenAddr
  74. }
  75. localAddr = ip.String()
  76. }
  77. }
  78. nr, err := c.newNodeRunner(nodeStartConfig{
  79. forceNewCluster: req.ForceNewCluster,
  80. autolock: req.AutoLockManagers,
  81. LocalAddr: localAddr,
  82. ListenAddr: net.JoinHostPort(listenHost, listenPort),
  83. AdvertiseAddr: net.JoinHostPort(advertiseHost, advertisePort),
  84. DataPathAddr: dataPathAddr,
  85. availability: req.Availability,
  86. })
  87. if err != nil {
  88. return "", err
  89. }
  90. c.mu.Lock()
  91. c.nr = nr
  92. c.mu.Unlock()
  93. if err := <-nr.Ready(); err != nil {
  94. c.mu.Lock()
  95. c.nr = nil
  96. c.mu.Unlock()
  97. if !req.ForceNewCluster { // if failure on first attempt don't keep state
  98. if err := clearPersistentState(c.root); err != nil {
  99. return "", err
  100. }
  101. }
  102. return "", err
  103. }
  104. state := nr.State()
  105. if state.swarmNode == nil { // should never happen but protect from panic
  106. return "", errors.New("invalid cluster state for spec initialization")
  107. }
  108. if err := initClusterSpec(state.swarmNode, req.Spec); err != nil {
  109. return "", err
  110. }
  111. return state.NodeID(), nil
  112. }
  113. // Join makes current Cluster part of an existing swarm cluster.
  114. func (c *Cluster) Join(req types.JoinRequest) error {
  115. c.controlMutex.Lock()
  116. defer c.controlMutex.Unlock()
  117. c.mu.Lock()
  118. if c.nr != nil {
  119. c.mu.Unlock()
  120. return errSwarmExists
  121. }
  122. c.mu.Unlock()
  123. if err := validateAndSanitizeJoinRequest(&req); err != nil {
  124. return apierrors.NewBadRequestError(err)
  125. }
  126. listenHost, listenPort, err := resolveListenAddr(req.ListenAddr)
  127. if err != nil {
  128. return err
  129. }
  130. var advertiseAddr string
  131. if req.AdvertiseAddr != "" {
  132. advertiseHost, advertisePort, err := c.resolveAdvertiseAddr(req.AdvertiseAddr, listenPort)
  133. // For joining, we don't need to provide an advertise address,
  134. // since the remote side can detect it.
  135. if err == nil {
  136. advertiseAddr = net.JoinHostPort(advertiseHost, advertisePort)
  137. }
  138. }
  139. dataPathAddr, err := resolveDataPathAddr(req.DataPathAddr)
  140. if err != nil {
  141. return err
  142. }
  143. nr, err := c.newNodeRunner(nodeStartConfig{
  144. RemoteAddr: req.RemoteAddrs[0],
  145. ListenAddr: net.JoinHostPort(listenHost, listenPort),
  146. AdvertiseAddr: advertiseAddr,
  147. DataPathAddr: dataPathAddr,
  148. joinAddr: req.RemoteAddrs[0],
  149. joinToken: req.JoinToken,
  150. availability: req.Availability,
  151. })
  152. if err != nil {
  153. return err
  154. }
  155. c.mu.Lock()
  156. c.nr = nr
  157. c.mu.Unlock()
  158. select {
  159. case <-time.After(swarmConnectTimeout):
  160. return errSwarmJoinTimeoutReached
  161. case err := <-nr.Ready():
  162. if err != nil {
  163. c.mu.Lock()
  164. c.nr = nil
  165. c.mu.Unlock()
  166. if err := clearPersistentState(c.root); err != nil {
  167. return err
  168. }
  169. }
  170. return err
  171. }
  172. }
  173. // Inspect retrieves the configuration properties of a managed swarm cluster.
  174. func (c *Cluster) Inspect() (types.Swarm, error) {
  175. var swarm *swarmapi.Cluster
  176. if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
  177. s, err := getSwarm(ctx, state.controlClient)
  178. if err != nil {
  179. return err
  180. }
  181. swarm = s
  182. return nil
  183. }); err != nil {
  184. return types.Swarm{}, err
  185. }
  186. return convert.SwarmFromGRPC(*swarm), nil
  187. }
  188. // Update updates configuration of a managed swarm cluster.
  189. func (c *Cluster) Update(version uint64, spec types.Spec, flags types.UpdateFlags) error {
  190. return c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
  191. swarm, err := getSwarm(ctx, state.controlClient)
  192. if err != nil {
  193. return err
  194. }
  195. // In update, client should provide the complete spec of the swarm, including
  196. // Name and Labels. If a field is specified with 0 or nil, then the default value
  197. // will be used to swarmkit.
  198. clusterSpec, err := convert.SwarmSpecToGRPC(spec)
  199. if err != nil {
  200. return apierrors.NewBadRequestError(err)
  201. }
  202. _, err = state.controlClient.UpdateCluster(
  203. ctx,
  204. &swarmapi.UpdateClusterRequest{
  205. ClusterID: swarm.ID,
  206. Spec: &clusterSpec,
  207. ClusterVersion: &swarmapi.Version{
  208. Index: version,
  209. },
  210. Rotation: swarmapi.KeyRotation{
  211. WorkerJoinToken: flags.RotateWorkerToken,
  212. ManagerJoinToken: flags.RotateManagerToken,
  213. ManagerUnlockKey: flags.RotateManagerUnlockKey,
  214. },
  215. },
  216. )
  217. return err
  218. })
  219. }
  220. // GetUnlockKey returns the unlock key for the swarm.
  221. func (c *Cluster) GetUnlockKey() (string, error) {
  222. var resp *swarmapi.GetUnlockKeyResponse
  223. if err := c.lockedManagerAction(func(ctx context.Context, state nodeState) error {
  224. client := swarmapi.NewCAClient(state.grpcConn)
  225. r, err := client.GetUnlockKey(ctx, &swarmapi.GetUnlockKeyRequest{})
  226. if err != nil {
  227. return err
  228. }
  229. resp = r
  230. return nil
  231. }); err != nil {
  232. return "", err
  233. }
  234. if len(resp.UnlockKey) == 0 {
  235. // no key
  236. return "", nil
  237. }
  238. return encryption.HumanReadableKey(resp.UnlockKey), nil
  239. }
  240. // UnlockSwarm provides a key to decrypt data that is encrypted at rest.
  241. func (c *Cluster) UnlockSwarm(req types.UnlockRequest) error {
  242. c.controlMutex.Lock()
  243. defer c.controlMutex.Unlock()
  244. c.mu.RLock()
  245. state := c.currentNodeState()
  246. if !state.IsActiveManager() {
  247. // when manager is not active,
  248. // unless it is locked, otherwise return error.
  249. if err := c.errNoManager(state); err != errSwarmLocked {
  250. c.mu.RUnlock()
  251. return err
  252. }
  253. } else {
  254. // when manager is active, return an error of "not locked"
  255. c.mu.RUnlock()
  256. return errors.New("swarm is not locked")
  257. }
  258. // only when swarm is locked, code running reaches here
  259. nr := c.nr
  260. c.mu.RUnlock()
  261. key, err := encryption.ParseHumanReadableKey(req.UnlockKey)
  262. if err != nil {
  263. return err
  264. }
  265. config := nr.config
  266. config.lockKey = key
  267. if err := nr.Stop(); err != nil {
  268. return err
  269. }
  270. nr, err = c.newNodeRunner(config)
  271. if err != nil {
  272. return err
  273. }
  274. c.mu.Lock()
  275. c.nr = nr
  276. c.mu.Unlock()
  277. if err := <-nr.Ready(); err != nil {
  278. if errors.Cause(err) == errSwarmLocked {
  279. return errors.New("swarm could not be unlocked: invalid key provided")
  280. }
  281. return fmt.Errorf("swarm component could not be started: %v", err)
  282. }
  283. return nil
  284. }
  285. // Leave shuts down Cluster and removes current state.
  286. func (c *Cluster) Leave(force bool) error {
  287. c.controlMutex.Lock()
  288. defer c.controlMutex.Unlock()
  289. c.mu.Lock()
  290. nr := c.nr
  291. if nr == nil {
  292. c.mu.Unlock()
  293. return errNoSwarm
  294. }
  295. state := c.currentNodeState()
  296. c.mu.Unlock()
  297. if errors.Cause(state.err) == errSwarmLocked && !force {
  298. // leave a locked swarm without --force is not allowed
  299. return errors.New("Swarm is encrypted and locked. Please unlock it first or use `--force` to ignore this message.")
  300. }
  301. if state.IsManager() && !force {
  302. msg := "You are attempting to leave the swarm on a node that is participating as a manager. "
  303. if state.IsActiveManager() {
  304. active, reachable, unreachable, err := managerStats(state.controlClient, state.NodeID())
  305. if err == nil {
  306. if active && removingManagerCausesLossOfQuorum(reachable, unreachable) {
  307. if isLastManager(reachable, unreachable) {
  308. msg += "Removing the last manager erases all current state of the swarm. Use `--force` to ignore this message. "
  309. return errors.New(msg)
  310. }
  311. msg += fmt.Sprintf("Removing this node leaves %v managers out of %v. Without a Raft quorum your swarm will be inaccessible. ", reachable-1, reachable+unreachable)
  312. }
  313. }
  314. } else {
  315. msg += "Doing so may lose the consensus of your cluster. "
  316. }
  317. msg += "The only way to restore a swarm that has lost consensus is to reinitialize it with `--force-new-cluster`. Use `--force` to suppress this message."
  318. return errors.New(msg)
  319. }
  320. // release readers in here
  321. if err := nr.Stop(); err != nil {
  322. logrus.Errorf("failed to shut down cluster node: %v", err)
  323. signal.DumpStacks("")
  324. return err
  325. }
  326. c.mu.Lock()
  327. c.nr = nil
  328. c.mu.Unlock()
  329. if nodeID := state.NodeID(); nodeID != "" {
  330. nodeContainers, err := c.listContainerForNode(nodeID)
  331. if err != nil {
  332. return err
  333. }
  334. for _, id := range nodeContainers {
  335. if err := c.config.Backend.ContainerRm(id, &apitypes.ContainerRmConfig{ForceRemove: true}); err != nil {
  336. logrus.Errorf("error removing %v: %v", id, err)
  337. }
  338. }
  339. }
  340. // todo: cleanup optional?
  341. if err := clearPersistentState(c.root); err != nil {
  342. return err
  343. }
  344. c.config.Backend.DaemonLeavesCluster()
  345. return nil
  346. }
  347. // Info returns information about the current cluster state.
  348. func (c *Cluster) Info() types.Info {
  349. info := types.Info{
  350. NodeAddr: c.GetAdvertiseAddress(),
  351. }
  352. c.mu.RLock()
  353. defer c.mu.RUnlock()
  354. state := c.currentNodeState()
  355. info.LocalNodeState = state.status
  356. if state.err != nil {
  357. info.Error = state.err.Error()
  358. }
  359. ctx, cancel := c.getRequestContext()
  360. defer cancel()
  361. if state.IsActiveManager() {
  362. info.ControlAvailable = true
  363. swarm, err := c.Inspect()
  364. if err != nil {
  365. info.Error = err.Error()
  366. }
  367. info.Cluster = &swarm.ClusterInfo
  368. if r, err := state.controlClient.ListNodes(ctx, &swarmapi.ListNodesRequest{}); err != nil {
  369. info.Error = err.Error()
  370. } else {
  371. info.Nodes = len(r.Nodes)
  372. for _, n := range r.Nodes {
  373. if n.ManagerStatus != nil {
  374. info.Managers = info.Managers + 1
  375. }
  376. }
  377. }
  378. }
  379. if state.swarmNode != nil {
  380. for _, r := range state.swarmNode.Remotes() {
  381. info.RemoteManagers = append(info.RemoteManagers, types.Peer{NodeID: r.NodeID, Addr: r.Addr})
  382. }
  383. info.NodeID = state.swarmNode.NodeID()
  384. }
  385. return info
  386. }
  387. func validateAndSanitizeInitRequest(req *types.InitRequest) error {
  388. var err error
  389. req.ListenAddr, err = validateAddr(req.ListenAddr)
  390. if err != nil {
  391. return fmt.Errorf("invalid ListenAddr %q: %v", req.ListenAddr, err)
  392. }
  393. if req.Spec.Annotations.Name == "" {
  394. req.Spec.Annotations.Name = "default"
  395. } else if req.Spec.Annotations.Name != "default" {
  396. return errors.New(`swarm spec must be named "default"`)
  397. }
  398. return nil
  399. }
  400. func validateAndSanitizeJoinRequest(req *types.JoinRequest) error {
  401. var err error
  402. req.ListenAddr, err = validateAddr(req.ListenAddr)
  403. if err != nil {
  404. return fmt.Errorf("invalid ListenAddr %q: %v", req.ListenAddr, err)
  405. }
  406. if len(req.RemoteAddrs) == 0 {
  407. return errors.New("at least 1 RemoteAddr is required to join")
  408. }
  409. for i := range req.RemoteAddrs {
  410. req.RemoteAddrs[i], err = validateAddr(req.RemoteAddrs[i])
  411. if err != nil {
  412. return fmt.Errorf("invalid remoteAddr %q: %v", req.RemoteAddrs[i], err)
  413. }
  414. }
  415. return nil
  416. }
  417. func validateAddr(addr string) (string, error) {
  418. if addr == "" {
  419. return addr, errors.New("invalid empty address")
  420. }
  421. newaddr, err := opts.ParseTCPAddr(addr, defaultAddr)
  422. if err != nil {
  423. return addr, nil
  424. }
  425. return strings.TrimPrefix(newaddr, "tcp://"), nil
  426. }
  427. func initClusterSpec(node *swarmnode.Node, spec types.Spec) error {
  428. ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
  429. for conn := range node.ListenControlSocket(ctx) {
  430. if ctx.Err() != nil {
  431. return ctx.Err()
  432. }
  433. if conn != nil {
  434. client := swarmapi.NewControlClient(conn)
  435. var cluster *swarmapi.Cluster
  436. for i := 0; ; i++ {
  437. lcr, err := client.ListClusters(ctx, &swarmapi.ListClustersRequest{})
  438. if err != nil {
  439. return fmt.Errorf("error on listing clusters: %v", err)
  440. }
  441. if len(lcr.Clusters) == 0 {
  442. if i < 10 {
  443. time.Sleep(200 * time.Millisecond)
  444. continue
  445. }
  446. return errors.New("empty list of clusters was returned")
  447. }
  448. cluster = lcr.Clusters[0]
  449. break
  450. }
  451. // In init, we take the initial default values from swarmkit, and merge
  452. // any non nil or 0 value from spec to GRPC spec. This will leave the
  453. // default value alone.
  454. // Note that this is different from Update(), as in Update() we expect
  455. // user to specify the complete spec of the cluster (as they already know
  456. // the existing one and knows which field to update)
  457. clusterSpec, err := convert.MergeSwarmSpecToGRPC(spec, cluster.Spec)
  458. if err != nil {
  459. return fmt.Errorf("error updating cluster settings: %v", err)
  460. }
  461. _, err = client.UpdateCluster(ctx, &swarmapi.UpdateClusterRequest{
  462. ClusterID: cluster.ID,
  463. ClusterVersion: &cluster.Meta.Version,
  464. Spec: &clusterSpec,
  465. })
  466. if err != nil {
  467. return fmt.Errorf("error updating cluster settings: %v", err)
  468. }
  469. return nil
  470. }
  471. }
  472. return ctx.Err()
  473. }
  474. func (c *Cluster) listContainerForNode(nodeID string) ([]string, error) {
  475. var ids []string
  476. filters := filters.NewArgs()
  477. filters.Add("label", fmt.Sprintf("com.docker.swarm.node.id=%s", nodeID))
  478. containers, err := c.config.Backend.Containers(&apitypes.ContainerListOptions{
  479. Filters: filters,
  480. })
  481. if err != nil {
  482. return []string{}, err
  483. }
  484. for _, c := range containers {
  485. ids = append(ids, c.ID)
  486. }
  487. return ids, nil
  488. }