mesh.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810
  1. // Copyright 2019 the Kilo authors
  2. //
  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. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package mesh
  15. import (
  16. "bytes"
  17. "fmt"
  18. "io/ioutil"
  19. "net"
  20. "os"
  21. "sync"
  22. "time"
  23. "github.com/go-kit/kit/log"
  24. "github.com/go-kit/kit/log/level"
  25. "github.com/prometheus/client_golang/prometheus"
  26. "github.com/vishvananda/netlink"
  27. "github.com/squat/kilo/pkg/iproute"
  28. "github.com/squat/kilo/pkg/iptables"
  29. "github.com/squat/kilo/pkg/route"
  30. "github.com/squat/kilo/pkg/wireguard"
  31. )
  32. const resyncPeriod = 30 * time.Second
  33. const (
  34. // KiloPath is the directory where Kilo stores its configuration.
  35. KiloPath = "/var/lib/kilo"
  36. // PrivateKeyPath is the filepath where the WireGuard private key is stored.
  37. PrivateKeyPath = KiloPath + "/key"
  38. // ConfPath is the filepath where the WireGuard configuration is stored.
  39. ConfPath = KiloPath + "/conf"
  40. // DefaultKiloPort is the default UDP port Kilo uses.
  41. DefaultKiloPort = 51820
  42. // DefaultCNIPath is the default path to the CNI config file.
  43. DefaultCNIPath = "/etc/cni/net.d/10-kilo.conflist"
  44. )
  45. // Granularity represents the abstraction level at which the network
  46. // should be meshed.
  47. type Granularity string
  48. // Encapsulate identifies what packets within a location should
  49. // be encapsulated.
  50. type Encapsulate string
  51. const (
  52. // LogicalGranularity indicates that the network should create
  53. // a mesh between logical locations, e.g. data-centers, but not between
  54. // all nodes within a single location.
  55. LogicalGranularity Granularity = "location"
  56. // FullGranularity indicates that the network should create
  57. // a mesh between every node.
  58. FullGranularity Granularity = "full"
  59. // NeverEncapsulate indicates that no packets within a location
  60. // should be encapsulated.
  61. NeverEncapsulate Encapsulate = "never"
  62. // CrossSubnetEncapsulate indicates that only packets that
  63. // traverse subnets within a location should be encapsulated.
  64. CrossSubnetEncapsulate Encapsulate = "crosssubnet"
  65. // AlwaysEncapsulate indicates that all packets within a location
  66. // should be encapsulated.
  67. AlwaysEncapsulate Encapsulate = "always"
  68. )
  69. // Node represents a node in the network.
  70. type Node struct {
  71. ExternalIP *net.IPNet
  72. Key []byte
  73. InternalIP *net.IPNet
  74. // LastSeen is a Unix time for the last time
  75. // the node confirmed it was live.
  76. LastSeen int64
  77. // Leader is a suggestion to Kilo that
  78. // the node wants to lead its segment.
  79. Leader bool
  80. Location string
  81. Name string
  82. Subnet *net.IPNet
  83. }
  84. // Ready indicates whether or not the node is ready.
  85. func (n *Node) Ready() bool {
  86. return n != nil && n.ExternalIP != nil && n.Key != nil && n.InternalIP != nil && n.Subnet != nil && time.Now().Unix()-n.LastSeen < int64(resyncPeriod)*2/int64(time.Second)
  87. }
  88. // Peer represents a peer in the network.
  89. type Peer struct {
  90. wireguard.Peer
  91. Name string
  92. }
  93. // Ready indicates whether or not the peer is ready.
  94. func (p *Peer) Ready() bool {
  95. return p != nil && p.AllowedIPs != nil && len(p.AllowedIPs) != 0 && p.PublicKey != nil
  96. }
  97. // EventType describes what kind of an action an event represents.
  98. type EventType string
  99. const (
  100. // AddEvent represents an action where an item was added.
  101. AddEvent EventType = "add"
  102. // DeleteEvent represents an action where an item was removed.
  103. DeleteEvent EventType = "delete"
  104. // UpdateEvent represents an action where an item was updated.
  105. UpdateEvent EventType = "update"
  106. )
  107. // NodeEvent represents an event concerning a node in the cluster.
  108. type NodeEvent struct {
  109. Type EventType
  110. Node *Node
  111. Old *Node
  112. }
  113. // PeerEvent represents an event concerning a peer in the cluster.
  114. type PeerEvent struct {
  115. Type EventType
  116. Peer *Peer
  117. Old *Peer
  118. }
  119. // Backend can create clients for all of the
  120. // primitive types that Kilo deals with, namely:
  121. // * nodes; and
  122. // * peers.
  123. type Backend interface {
  124. Nodes() NodeBackend
  125. Peers() PeerBackend
  126. }
  127. // NodeBackend can get nodes by name, init itself,
  128. // list the nodes that should be meshed,
  129. // set Kilo properties for a node,
  130. // clean up any changes applied to the backend,
  131. // and watch for changes to nodes.
  132. type NodeBackend interface {
  133. CleanUp(string) error
  134. Get(string) (*Node, error)
  135. Init(<-chan struct{}) error
  136. List() ([]*Node, error)
  137. Set(string, *Node) error
  138. Watch() <-chan *NodeEvent
  139. }
  140. // PeerBackend can get peers by name, init itself,
  141. // list the peers that should be in the mesh,
  142. // set fields for a peer,
  143. // clean up any changes applied to the backend,
  144. // and watch for changes to peers.
  145. type PeerBackend interface {
  146. CleanUp(string) error
  147. Get(string) (*Peer, error)
  148. Init(<-chan struct{}) error
  149. List() ([]*Peer, error)
  150. Set(string, *Peer) error
  151. Watch() <-chan *PeerEvent
  152. }
  153. // Mesh is able to create Kilo network meshes.
  154. type Mesh struct {
  155. Backend
  156. cni bool
  157. cniPath string
  158. encapsulate Encapsulate
  159. externalIP *net.IPNet
  160. granularity Granularity
  161. hostname string
  162. internalIP *net.IPNet
  163. ipTables *iptables.Controller
  164. kiloIface int
  165. key []byte
  166. local bool
  167. port uint32
  168. priv []byte
  169. privIface int
  170. pub []byte
  171. pubIface int
  172. stop chan struct{}
  173. subnet *net.IPNet
  174. table *route.Table
  175. tunlIface int
  176. // nodes and peers are mutable fields in the struct
  177. // and needs to be guarded.
  178. nodes map[string]*Node
  179. peers map[string]*Peer
  180. mu sync.Mutex
  181. errorCounter *prometheus.CounterVec
  182. nodesGuage prometheus.Gauge
  183. peersGuage prometheus.Gauge
  184. reconcileCounter prometheus.Counter
  185. logger log.Logger
  186. }
  187. // New returns a new Mesh instance.
  188. func New(backend Backend, encapsulate Encapsulate, granularity Granularity, hostname string, port uint32, subnet *net.IPNet, local, cni bool, cniPath string, logger log.Logger) (*Mesh, error) {
  189. if err := os.MkdirAll(KiloPath, 0700); err != nil {
  190. return nil, fmt.Errorf("failed to create directory to store configuration: %v", err)
  191. }
  192. private, err := ioutil.ReadFile(PrivateKeyPath)
  193. private = bytes.Trim(private, "\n")
  194. if err != nil {
  195. level.Warn(logger).Log("msg", "no private key found on disk; generating one now")
  196. if private, err = wireguard.GenKey(); err != nil {
  197. return nil, err
  198. }
  199. }
  200. public, err := wireguard.PubKey(private)
  201. if err != nil {
  202. return nil, err
  203. }
  204. if err := ioutil.WriteFile(PrivateKeyPath, private, 0600); err != nil {
  205. return nil, fmt.Errorf("failed to write private key to disk: %v", err)
  206. }
  207. privateIP, publicIP, err := getIP(hostname)
  208. if err != nil {
  209. return nil, fmt.Errorf("failed to find public IP: %v", err)
  210. }
  211. ifaces, err := interfacesForIP(privateIP)
  212. if err != nil {
  213. return nil, fmt.Errorf("failed to find interface for private IP: %v", err)
  214. }
  215. privIface := ifaces[0].Index
  216. ifaces, err = interfacesForIP(publicIP)
  217. if err != nil {
  218. return nil, fmt.Errorf("failed to find interface for public IP: %v", err)
  219. }
  220. pubIface := ifaces[0].Index
  221. kiloIface, err := wireguard.New("kilo")
  222. if err != nil {
  223. return nil, fmt.Errorf("failed to create WireGuard interface: %v", err)
  224. }
  225. var tunlIface int
  226. if encapsulate != NeverEncapsulate {
  227. if tunlIface, err = iproute.NewIPIP(privIface); err != nil {
  228. return nil, fmt.Errorf("failed to create tunnel interface: %v", err)
  229. }
  230. if err := iproute.Set(tunlIface, true); err != nil {
  231. return nil, fmt.Errorf("failed to set tunnel interface up: %v", err)
  232. }
  233. }
  234. level.Debug(logger).Log("msg", fmt.Sprintf("using %s as the private IP address", privateIP.String()))
  235. level.Debug(logger).Log("msg", fmt.Sprintf("using %s as the public IP address", publicIP.String()))
  236. ipTables, err := iptables.New(len(subnet.IP))
  237. if err != nil {
  238. return nil, fmt.Errorf("failed to IP tables controller: %v", err)
  239. }
  240. return &Mesh{
  241. Backend: backend,
  242. cni: cni,
  243. cniPath: cniPath,
  244. encapsulate: encapsulate,
  245. externalIP: publicIP,
  246. granularity: granularity,
  247. hostname: hostname,
  248. internalIP: privateIP,
  249. ipTables: ipTables,
  250. kiloIface: kiloIface,
  251. nodes: make(map[string]*Node),
  252. peers: make(map[string]*Peer),
  253. port: port,
  254. priv: private,
  255. privIface: privIface,
  256. pub: public,
  257. pubIface: pubIface,
  258. local: local,
  259. stop: make(chan struct{}),
  260. subnet: subnet,
  261. table: route.NewTable(),
  262. tunlIface: tunlIface,
  263. errorCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
  264. Name: "kilo_errors_total",
  265. Help: "Number of errors that occurred while administering the mesh.",
  266. }, []string{"event"}),
  267. nodesGuage: prometheus.NewGauge(prometheus.GaugeOpts{
  268. Name: "kilo_nodes",
  269. Help: "Number of nodes in the mesh.",
  270. }),
  271. peersGuage: prometheus.NewGauge(prometheus.GaugeOpts{
  272. Name: "kilo_peers",
  273. Help: "Number of peers in the mesh.",
  274. }),
  275. reconcileCounter: prometheus.NewCounter(prometheus.CounterOpts{
  276. Name: "kilo_reconciles_total",
  277. Help: "Number of reconciliation attempts.",
  278. }),
  279. logger: logger,
  280. }, nil
  281. }
  282. // Run starts the mesh.
  283. func (m *Mesh) Run() error {
  284. if err := m.Nodes().Init(m.stop); err != nil {
  285. return fmt.Errorf("failed to initialize node backend: %v", err)
  286. }
  287. if err := m.Peers().Init(m.stop); err != nil {
  288. return fmt.Errorf("failed to initialize peer backend: %v", err)
  289. }
  290. ipTablesErrors, err := m.ipTables.Run(m.stop)
  291. if err != nil {
  292. return fmt.Errorf("failed to watch for IP tables updates: %v", err)
  293. }
  294. routeErrors, err := m.table.Run(m.stop)
  295. if err != nil {
  296. return fmt.Errorf("failed to watch for route table updates: %v", err)
  297. }
  298. go func() {
  299. for {
  300. var err error
  301. select {
  302. case err = <-ipTablesErrors:
  303. case err = <-routeErrors:
  304. case <-m.stop:
  305. return
  306. }
  307. if err != nil {
  308. level.Error(m.logger).Log("error", err)
  309. m.errorCounter.WithLabelValues("run").Inc()
  310. }
  311. }
  312. }()
  313. defer m.cleanUp()
  314. t := time.NewTimer(resyncPeriod)
  315. nw := m.Nodes().Watch()
  316. pw := m.Peers().Watch()
  317. var ne *NodeEvent
  318. var pe *PeerEvent
  319. for {
  320. select {
  321. case ne = <-nw:
  322. m.syncNodes(ne)
  323. case pe = <-pw:
  324. m.syncPeers(pe)
  325. case <-t.C:
  326. m.checkIn()
  327. if m.cni {
  328. m.updateCNIConfig()
  329. }
  330. m.syncEndpoints()
  331. m.applyTopology()
  332. t.Reset(resyncPeriod)
  333. case <-m.stop:
  334. return nil
  335. }
  336. }
  337. }
  338. // WireGuard updates the endpoints of peers to match the
  339. // last place a valid packet was received from.
  340. // Periodically we need to syncronize the endpoints
  341. // of peers in the backend to match the WireGuard configuration.
  342. func (m *Mesh) syncEndpoints() {
  343. link, err := linkByIndex(m.kiloIface)
  344. if err != nil {
  345. level.Error(m.logger).Log("error", err)
  346. m.errorCounter.WithLabelValues("endpoints").Inc()
  347. return
  348. }
  349. conf, err := wireguard.ShowConf(link.Attrs().Name)
  350. if err != nil {
  351. level.Error(m.logger).Log("error", err)
  352. m.errorCounter.WithLabelValues("endpoints").Inc()
  353. return
  354. }
  355. m.mu.Lock()
  356. defer m.mu.Unlock()
  357. c := wireguard.Parse(conf)
  358. var key string
  359. var tmp *Peer
  360. for i := range c.Peers {
  361. // Peers are indexed by public key.
  362. key = string(c.Peers[i].PublicKey)
  363. if p, ok := m.peers[key]; ok {
  364. tmp = &Peer{
  365. Name: p.Name,
  366. Peer: *c.Peers[i],
  367. }
  368. if !peersAreEqual(tmp, p) {
  369. p.Endpoint = tmp.Endpoint
  370. if err := m.Peers().Set(p.Name, p); err != nil {
  371. level.Error(m.logger).Log("error", err)
  372. m.errorCounter.WithLabelValues("endpoints").Inc()
  373. }
  374. }
  375. }
  376. }
  377. }
  378. func (m *Mesh) syncNodes(e *NodeEvent) {
  379. logger := log.With(m.logger, "event", e.Type)
  380. level.Debug(logger).Log("msg", "syncing nodes", "event", e.Type)
  381. if isSelf(m.hostname, e.Node) {
  382. level.Debug(logger).Log("msg", "processing local node", "node", e.Node)
  383. m.handleLocal(e.Node)
  384. return
  385. }
  386. var diff bool
  387. m.mu.Lock()
  388. if !e.Node.Ready() {
  389. level.Debug(logger).Log("msg", "received incomplete node", "node", e.Node)
  390. // An existing node is no longer valid
  391. // so remove it from the mesh.
  392. if _, ok := m.nodes[e.Node.Name]; ok {
  393. level.Info(logger).Log("msg", "node is no longer ready", "node", e.Node)
  394. diff = true
  395. }
  396. } else {
  397. switch e.Type {
  398. case AddEvent:
  399. fallthrough
  400. case UpdateEvent:
  401. if !nodesAreEqual(m.nodes[e.Node.Name], e.Node) {
  402. diff = true
  403. }
  404. // Even if the nodes are the same,
  405. // overwrite the old node to update the timestamp.
  406. m.nodes[e.Node.Name] = e.Node
  407. case DeleteEvent:
  408. delete(m.nodes, e.Node.Name)
  409. diff = true
  410. }
  411. }
  412. m.mu.Unlock()
  413. if diff {
  414. level.Info(logger).Log("node", e.Node)
  415. m.applyTopology()
  416. }
  417. }
  418. func (m *Mesh) syncPeers(e *PeerEvent) {
  419. logger := log.With(m.logger, "event", e.Type)
  420. level.Debug(logger).Log("msg", "syncing peers", "event", e.Type)
  421. var diff bool
  422. m.mu.Lock()
  423. // Peers are indexed by public key.
  424. key := string(e.Peer.PublicKey)
  425. if !e.Peer.Ready() {
  426. level.Debug(logger).Log("msg", "received incomplete peer", "peer", e.Peer)
  427. // An existing peer is no longer valid
  428. // so remove it from the mesh.
  429. if _, ok := m.peers[key]; ok {
  430. level.Info(logger).Log("msg", "peer is no longer ready", "peer", e.Peer)
  431. diff = true
  432. }
  433. } else {
  434. switch e.Type {
  435. case AddEvent:
  436. fallthrough
  437. case UpdateEvent:
  438. if e.Old != nil && key != string(e.Old.PublicKey) {
  439. delete(m.peers, string(e.Old.PublicKey))
  440. diff = true
  441. }
  442. if !peersAreEqual(m.peers[key], e.Peer) {
  443. m.peers[key] = e.Peer
  444. diff = true
  445. }
  446. case DeleteEvent:
  447. delete(m.peers, key)
  448. diff = true
  449. }
  450. }
  451. m.mu.Unlock()
  452. if diff {
  453. level.Info(logger).Log("peer", e.Peer)
  454. m.applyTopology()
  455. }
  456. }
  457. // checkIn will try to update the local node's LastSeen timestamp
  458. // in the backend.
  459. func (m *Mesh) checkIn() {
  460. m.mu.Lock()
  461. defer m.mu.Unlock()
  462. n := m.nodes[m.hostname]
  463. if n == nil {
  464. level.Debug(m.logger).Log("msg", "no local node found in backend")
  465. return
  466. }
  467. oldTime := n.LastSeen
  468. n.LastSeen = time.Now().Unix()
  469. if err := m.Nodes().Set(m.hostname, n); err != nil {
  470. level.Error(m.logger).Log("error", fmt.Sprintf("failed to set local node: %v", err), "node", n)
  471. m.errorCounter.WithLabelValues("checkin").Inc()
  472. // Revert time.
  473. n.LastSeen = oldTime
  474. return
  475. }
  476. level.Debug(m.logger).Log("msg", "successfully checked in local node in backend")
  477. }
  478. func (m *Mesh) handleLocal(n *Node) {
  479. // Allow the external IP to be overridden.
  480. if n.ExternalIP == nil {
  481. n.ExternalIP = m.externalIP
  482. }
  483. // Compare the given node to the calculated local node.
  484. // Take leader, location, and subnet from the argument, as these
  485. // are not determined by kilo.
  486. local := &Node{
  487. ExternalIP: n.ExternalIP,
  488. Key: m.pub,
  489. InternalIP: m.internalIP,
  490. LastSeen: time.Now().Unix(),
  491. Leader: n.Leader,
  492. Location: n.Location,
  493. Name: m.hostname,
  494. Subnet: n.Subnet,
  495. }
  496. if !nodesAreEqual(n, local) {
  497. level.Debug(m.logger).Log("msg", "local node differs from backend")
  498. if err := m.Nodes().Set(m.hostname, local); err != nil {
  499. level.Error(m.logger).Log("error", fmt.Sprintf("failed to set local node: %v", err), "node", local)
  500. m.errorCounter.WithLabelValues("local").Inc()
  501. return
  502. }
  503. level.Debug(m.logger).Log("msg", "successfully reconciled local node against backend")
  504. }
  505. m.mu.Lock()
  506. n = m.nodes[m.hostname]
  507. if n == nil {
  508. n = &Node{}
  509. }
  510. m.mu.Unlock()
  511. if !nodesAreEqual(n, local) {
  512. m.mu.Lock()
  513. m.nodes[local.Name] = local
  514. m.mu.Unlock()
  515. m.applyTopology()
  516. }
  517. }
  518. func (m *Mesh) applyTopology() {
  519. m.reconcileCounter.Inc()
  520. m.mu.Lock()
  521. defer m.mu.Unlock()
  522. // Ensure only ready nodes are considered.
  523. nodes := make(map[string]*Node)
  524. var readyNodes float64
  525. for k := range m.nodes {
  526. if !m.nodes[k].Ready() {
  527. continue
  528. }
  529. nodes[k] = m.nodes[k]
  530. readyNodes++
  531. }
  532. // Ensure only ready nodes are considered.
  533. peers := make(map[string]*Peer)
  534. var readyPeers float64
  535. for k := range m.peers {
  536. if !m.peers[k].Ready() {
  537. continue
  538. }
  539. peers[k] = m.peers[k]
  540. readyPeers++
  541. }
  542. m.nodesGuage.Set(readyNodes)
  543. m.peersGuage.Set(readyPeers)
  544. // We cannot do anything with the topology until the local node is available.
  545. if nodes[m.hostname] == nil {
  546. return
  547. }
  548. t, err := NewTopology(nodes, peers, m.granularity, m.hostname, m.port, m.priv, m.subnet)
  549. if err != nil {
  550. level.Error(m.logger).Log("error", err)
  551. m.errorCounter.WithLabelValues("apply").Inc()
  552. return
  553. }
  554. conf := t.Conf()
  555. buf, err := conf.Bytes()
  556. if err != nil {
  557. level.Error(m.logger).Log("error", err)
  558. m.errorCounter.WithLabelValues("apply").Inc()
  559. }
  560. if err := ioutil.WriteFile(ConfPath, buf, 0600); err != nil {
  561. level.Error(m.logger).Log("error", err)
  562. m.errorCounter.WithLabelValues("apply").Inc()
  563. return
  564. }
  565. rules := iptables.ForwardRules(m.subnet)
  566. var peerCIDRs []*net.IPNet
  567. for _, p := range peers {
  568. rules = append(rules, iptables.ForwardRules(p.AllowedIPs...)...)
  569. peerCIDRs = append(peerCIDRs, p.AllowedIPs...)
  570. }
  571. rules = append(rules, iptables.MasqueradeRules(m.subnet, oneAddressCIDR(t.privateIP.IP), nodes[m.hostname].Subnet, t.RemoteSubnets(), peerCIDRs)...)
  572. // If we are handling local routes, ensure the local
  573. // tunnel has an IP address and IPIP traffic is allowed.
  574. if m.encapsulate != NeverEncapsulate && m.local {
  575. var cidrs []*net.IPNet
  576. for _, s := range t.segments {
  577. if s.location == nodes[m.hostname].Location {
  578. for i := range s.privateIPs {
  579. cidrs = append(cidrs, oneAddressCIDR(s.privateIPs[i]))
  580. }
  581. break
  582. }
  583. }
  584. rules = append(rules, iptables.EncapsulateRules(cidrs)...)
  585. // If we are handling local routes, ensure the local
  586. // tunnel has an IP address.
  587. if err := iproute.SetAddress(m.tunlIface, oneAddressCIDR(newAllocator(*nodes[m.hostname].Subnet).next().IP)); err != nil {
  588. level.Error(m.logger).Log("error", err)
  589. m.errorCounter.WithLabelValues("apply").Inc()
  590. return
  591. }
  592. }
  593. if err := m.ipTables.Set(rules); err != nil {
  594. level.Error(m.logger).Log("error", err)
  595. m.errorCounter.WithLabelValues("apply").Inc()
  596. return
  597. }
  598. if t.leader {
  599. if err := iproute.SetAddress(m.kiloIface, t.wireGuardCIDR); err != nil {
  600. level.Error(m.logger).Log("error", err)
  601. m.errorCounter.WithLabelValues("apply").Inc()
  602. return
  603. }
  604. link, err := linkByIndex(m.kiloIface)
  605. if err != nil {
  606. level.Error(m.logger).Log("error", err)
  607. m.errorCounter.WithLabelValues("apply").Inc()
  608. return
  609. }
  610. oldConf, err := wireguard.ShowConf(link.Attrs().Name)
  611. if err != nil {
  612. level.Error(m.logger).Log("error", err)
  613. m.errorCounter.WithLabelValues("apply").Inc()
  614. return
  615. }
  616. // Setting the WireGuard configuration interrupts existing connections
  617. // so only set the configuration if it has changed.
  618. equal := conf.Equal(wireguard.Parse(oldConf))
  619. if !equal {
  620. level.Info(m.logger).Log("msg", "WireGuard configurations are different")
  621. if err := wireguard.SetConf(link.Attrs().Name, ConfPath); err != nil {
  622. level.Error(m.logger).Log("error", err)
  623. m.errorCounter.WithLabelValues("apply").Inc()
  624. return
  625. }
  626. }
  627. if err := iproute.Set(m.kiloIface, true); err != nil {
  628. level.Error(m.logger).Log("error", err)
  629. m.errorCounter.WithLabelValues("apply").Inc()
  630. return
  631. }
  632. } else {
  633. level.Debug(m.logger).Log("msg", "local node is not the leader")
  634. if err := iproute.Set(m.kiloIface, false); err != nil {
  635. level.Error(m.logger).Log("error", err)
  636. m.errorCounter.WithLabelValues("apply").Inc()
  637. return
  638. }
  639. }
  640. // We need to add routes last since they may depend
  641. // on the WireGuard interface.
  642. routes := t.Routes(m.kiloIface, m.privIface, m.tunlIface, m.local, m.encapsulate)
  643. if err := m.table.Set(routes); err != nil {
  644. level.Error(m.logger).Log("error", err)
  645. m.errorCounter.WithLabelValues("apply").Inc()
  646. }
  647. }
  648. // RegisterMetrics registers Prometheus metrics on the given Prometheus
  649. // registerer.
  650. func (m *Mesh) RegisterMetrics(r prometheus.Registerer) {
  651. r.MustRegister(
  652. m.errorCounter,
  653. m.nodesGuage,
  654. m.peersGuage,
  655. m.reconcileCounter,
  656. )
  657. }
  658. // Stop stops the mesh.
  659. func (m *Mesh) Stop() {
  660. close(m.stop)
  661. }
  662. func (m *Mesh) cleanUp() {
  663. if err := m.ipTables.CleanUp(); err != nil {
  664. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up IP tables: %v", err))
  665. m.errorCounter.WithLabelValues("cleanUp").Inc()
  666. }
  667. if err := m.table.CleanUp(); err != nil {
  668. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up routes: %v", err))
  669. m.errorCounter.WithLabelValues("cleanUp").Inc()
  670. }
  671. if err := os.Remove(PrivateKeyPath); err != nil {
  672. level.Error(m.logger).Log("error", fmt.Sprintf("failed to delete private key: %v", err))
  673. m.errorCounter.WithLabelValues("cleanUp").Inc()
  674. }
  675. if err := os.Remove(ConfPath); err != nil {
  676. level.Error(m.logger).Log("error", fmt.Sprintf("failed to delete configuration file: %v", err))
  677. m.errorCounter.WithLabelValues("cleanUp").Inc()
  678. }
  679. if err := iproute.RemoveInterface(m.kiloIface); err != nil {
  680. level.Error(m.logger).Log("error", fmt.Sprintf("failed to remove WireGuard interface: %v", err))
  681. m.errorCounter.WithLabelValues("cleanUp").Inc()
  682. }
  683. if err := m.Nodes().CleanUp(m.hostname); err != nil {
  684. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up node backend: %v", err))
  685. m.errorCounter.WithLabelValues("cleanUp").Inc()
  686. }
  687. if err := m.Peers().CleanUp(m.hostname); err != nil {
  688. level.Error(m.logger).Log("error", fmt.Sprintf("failed to clean up peer backend: %v", err))
  689. m.errorCounter.WithLabelValues("cleanUp").Inc()
  690. }
  691. }
  692. func isSelf(hostname string, node *Node) bool {
  693. return node != nil && node.Name == hostname
  694. }
  695. func nodesAreEqual(a, b *Node) bool {
  696. if !(a != nil) == (b != nil) {
  697. return false
  698. }
  699. if a == b {
  700. return true
  701. }
  702. // Ignore LastSeen when comparing equality we want to check if the nodes are
  703. // equivalent. However, we do want to check if LastSeen has transitioned
  704. // between valid and invalid.
  705. return ipNetsEqual(a.ExternalIP, b.ExternalIP) && string(a.Key) == string(b.Key) && ipNetsEqual(a.InternalIP, b.InternalIP) && a.Leader == b.Leader && a.Location == b.Location && a.Name == b.Name && subnetsEqual(a.Subnet, b.Subnet) && a.Ready() == b.Ready()
  706. }
  707. func peersAreEqual(a, b *Peer) bool {
  708. if !(a != nil) == (b != nil) {
  709. return false
  710. }
  711. if a == b {
  712. return true
  713. }
  714. if !(a.Endpoint != nil) == (b.Endpoint != nil) {
  715. return false
  716. }
  717. if a.Endpoint != nil {
  718. if !a.Endpoint.IP.Equal(b.Endpoint.IP) || a.Endpoint.Port != b.Endpoint.Port {
  719. return false
  720. }
  721. }
  722. if len(a.AllowedIPs) != len(b.AllowedIPs) {
  723. return false
  724. }
  725. for i := range a.AllowedIPs {
  726. if !ipNetsEqual(a.AllowedIPs[i], b.AllowedIPs[i]) {
  727. return false
  728. }
  729. }
  730. return string(a.PublicKey) == string(b.PublicKey) && a.PersistentKeepalive == b.PersistentKeepalive
  731. }
  732. func ipNetsEqual(a, b *net.IPNet) bool {
  733. if a == nil && b == nil {
  734. return true
  735. }
  736. if (a != nil) != (b != nil) {
  737. return false
  738. }
  739. if a.Mask.String() != b.Mask.String() {
  740. return false
  741. }
  742. return a.IP.Equal(b.IP)
  743. }
  744. func subnetsEqual(a, b *net.IPNet) bool {
  745. if a == nil && b == nil {
  746. return true
  747. }
  748. if (a != nil) != (b != nil) {
  749. return false
  750. }
  751. if a.Mask.String() != b.Mask.String() {
  752. return false
  753. }
  754. if !a.Contains(b.IP) {
  755. return false
  756. }
  757. if !b.Contains(a.IP) {
  758. return false
  759. }
  760. return true
  761. }
  762. func linkByIndex(index int) (netlink.Link, error) {
  763. link, err := netlink.LinkByIndex(index)
  764. if err != nil {
  765. return nil, fmt.Errorf("failed to get interface: %v", err)
  766. }
  767. return link, nil
  768. }