123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498 |
- // Copyright 2015 flannel authors
- //
- // Licensed under the Apache License, Version 2.0 (the "License");
- // you may not use this file except in compliance with the License.
- // You may obtain a copy of the License at
- //
- // http://www.apache.org/licenses/LICENSE-2.0
- //
- // Unless required by applicable law or agreed to in writing, software
- // distributed under the License is distributed on an "AS IS" BASIS,
- // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- // See the License for the specific language governing permissions and
- // limitations under the License.
- package main
- import (
- "errors"
- "flag"
- "fmt"
- "net"
- "net/http"
- "os"
- "os/signal"
- "path/filepath"
- "regexp"
- "strconv"
- "strings"
- "syscall"
- "github.com/coreos/pkg/flagutil"
- log "github.com/golang/glog"
- "golang.org/x/net/context"
- "github.com/coreos/flannel/network"
- "github.com/coreos/flannel/pkg/ip"
- "github.com/coreos/flannel/subnet"
- "github.com/coreos/flannel/subnet/etcdv2"
- "github.com/coreos/flannel/subnet/kube"
- "github.com/coreos/flannel/version"
- "time"
- "github.com/joho/godotenv"
- // Backends need to be imported for their init() to get executed and them to register
- "github.com/coreos/flannel/backend"
- _ "github.com/coreos/flannel/backend/alivpc"
- _ "github.com/coreos/flannel/backend/alloc"
- _ "github.com/coreos/flannel/backend/awsvpc"
- _ "github.com/coreos/flannel/backend/extension"
- _ "github.com/coreos/flannel/backend/gce"
- _ "github.com/coreos/flannel/backend/hostgw"
- _ "github.com/coreos/flannel/backend/udp"
- _ "github.com/coreos/flannel/backend/vxlan"
- "github.com/coreos/go-systemd/daemon"
- )
- type CmdLineOpts struct {
- etcdEndpoints string
- etcdPrefix string
- etcdKeyfile string
- etcdCertfile string
- etcdCAFile string
- etcdUsername string
- etcdPassword string
- help bool
- version bool
- kubeSubnetMgr bool
- kubeApiUrl string
- kubeConfigFile string
- iface string
- ifaceRegex string
- ipMasq bool
- subnetFile string
- subnetDir string
- publicIP string
- subnetLeaseRenewMargin int
- healthzIP string
- healthzPort int
- }
- var (
- opts CmdLineOpts
- errInterrupted = errors.New("interrupted")
- errCanceled = errors.New("canceled")
- flannelFlags = flag.NewFlagSet("flannel", flag.ExitOnError)
- )
- func init() {
- flannelFlags.StringVar(&opts.etcdEndpoints, "etcd-endpoints", "http://127.0.0.1:4001,http://127.0.0.1:2379", "a comma-delimited list of etcd endpoints")
- flannelFlags.StringVar(&opts.etcdPrefix, "etcd-prefix", "/coreos.com/network", "etcd prefix")
- flannelFlags.StringVar(&opts.etcdKeyfile, "etcd-keyfile", "", "SSL key file used to secure etcd communication")
- flannelFlags.StringVar(&opts.etcdCertfile, "etcd-certfile", "", "SSL certification file used to secure etcd communication")
- flannelFlags.StringVar(&opts.etcdCAFile, "etcd-cafile", "", "SSL Certificate Authority file used to secure etcd communication")
- flannelFlags.StringVar(&opts.etcdUsername, "etcd-username", "", "username for BasicAuth to etcd")
- flannelFlags.StringVar(&opts.etcdPassword, "etcd-password", "", "password for BasicAuth to etcd")
- flannelFlags.StringVar(&opts.iface, "iface", "", "interface to use (IP or name) for inter-host communication")
- flannelFlags.StringVar(&opts.ifaceRegex, "iface-regex", "", "regex expression to match the first interface to use (IP or name) for inter-host communication. Skipped if the iface option is also specified")
- flannelFlags.StringVar(&opts.subnetFile, "subnet-file", "/run/flannel/subnet.env", "filename where env variables (subnet, MTU, ... ) will be written to")
- flannelFlags.StringVar(&opts.publicIP, "public-ip", "", "IP accessible by other nodes for inter-host communication")
- flannelFlags.IntVar(&opts.subnetLeaseRenewMargin, "subnet-lease-renew-margin", 60, "subnet lease renewal margin, in minutes.")
- flannelFlags.BoolVar(&opts.ipMasq, "ip-masq", false, "setup IP masquerade rule for traffic destined outside of overlay network")
- flannelFlags.BoolVar(&opts.kubeSubnetMgr, "kube-subnet-mgr", false, "contact the Kubernetes API for subnet assignment instead of etcd.")
- flannelFlags.StringVar(&opts.kubeApiUrl, "kube-api-url", "", "Kubernetes API server URL. Does not need to be specified if flannel is running in a pod.")
- flannelFlags.StringVar(&opts.kubeConfigFile, "kubeconfig-file", "", "kubeconfig file location. Does not need to be specified if flannel is running in a pod.")
- flannelFlags.BoolVar(&opts.version, "version", false, "print version and exit")
- flannelFlags.StringVar(&opts.healthzIP, "healthz-ip", "0.0.0.0", "the IP address for healthz server to listen")
- flannelFlags.IntVar(&opts.healthzPort, "healthz-port", 0, "the port for healthz server to listen(0 to disable)")
- // glog will log to tmp files by default. override so all entries
- // can flow into journald (if running under systemd)
- flag.Set("logtostderr", "true")
- // Only copy the non file logging options from glog
- copyFlag("v")
- copyFlag("vmodule")
- copyFlag("log_backtrace_at")
- // Define the usage function
- flannelFlags.Usage = usage
- // now parse command line args
- flannelFlags.Parse(os.Args[1:])
- }
- func copyFlag(name string) {
- flannelFlags.Var(flag.Lookup(name).Value, flag.Lookup(name).Name, flag.Lookup(name).Usage)
- }
- func usage() {
- fmt.Fprintf(os.Stderr, "Usage: %s [OPTION]...\n", os.Args[0])
- flannelFlags.PrintDefaults()
- os.Exit(0)
- }
- func newSubnetManager() (subnet.Manager, error) {
- if opts.kubeSubnetMgr {
- return kube.NewSubnetManager(opts.kubeApiUrl, opts.kubeConfigFile)
- }
- cfg := &etcdv2.EtcdConfig{
- Endpoints: strings.Split(opts.etcdEndpoints, ","),
- Keyfile: opts.etcdKeyfile,
- Certfile: opts.etcdCertfile,
- CAFile: opts.etcdCAFile,
- Prefix: opts.etcdPrefix,
- Username: opts.etcdUsername,
- Password: opts.etcdPassword,
- }
- // Attempt to renew the lease for the subnet specified in the subnetFile
- prevSubnet := ReadSubnetFromSubnetFile(opts.subnetFile)
- return etcdv2.NewLocalManager(cfg, prevSubnet)
- }
- func main() {
- if opts.version {
- fmt.Fprintln(os.Stderr, version.Version)
- os.Exit(0)
- }
- flagutil.SetFlagsFromEnv(flannelFlags, "FLANNELD")
- // Work out which interface to use
- extIface, err := LookupExtIface(opts.iface, opts.ifaceRegex)
- if err != nil {
- log.Error("Failed to find interface to use: ", err)
- os.Exit(1)
- }
- sm, err := newSubnetManager()
- if err != nil {
- log.Error("Failed to create SubnetManager: ", err)
- os.Exit(1)
- }
- log.Infof("Created subnet manager: %+v", sm)
- // Register for SIGINT and SIGTERM
- log.Info("Installing signal handlers")
- sigs := make(chan os.Signal, 1)
- signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)
- ctx, cancel := context.WithCancel(context.Background())
- go shutdown(sigs, cancel)
- if opts.healthzPort > 0 {
- go mustRunHealthz()
- }
- // Fetch the network config (i.e. what backend to use etc..).
- config, err := getConfig(ctx, sm)
- if err == errCanceled {
- exit()
- }
- // Create a backend manager then use it to create the backend and register the network with it.
- bm := backend.NewManager(ctx, sm, extIface)
- be, err := bm.GetBackend(config.BackendType)
- if err != nil {
- log.Errorf("Error fetching backend: %s", err)
- exit()
- }
- bn, err := be.RegisterNetwork(ctx, config)
- if err != nil {
- log.Errorf("Error registering network: %s", err)
- exit()
- }
- // Set up ipMasq if needed
- if opts.ipMasq {
- err = network.SetupIPMasq(config.Network, bn.Lease())
- if err != nil {
- // Continue, even though it failed.
- log.Errorf("Failed to set up IP Masquerade: %v", err)
- }
- defer func() {
- if err := network.TeardownIPMasq(config.Network, bn.Lease()); err != nil {
- log.Errorf("Failed to tear down IP Masquerade: %v", err)
- }
- }()
- }
- if err := WriteSubnetFile(opts.subnetFile, config.Network, opts.ipMasq, bn); err != nil {
- // Continue, even though it failed.
- log.Warningf("Failed to write subnet file: %s", err)
- } else {
- log.Infof("Wrote subnet file to %s", opts.subnetFile)
- }
- // Start "Running" the backend network. This will block until the context is done so run in another goroutine.
- go bn.Run(ctx)
- log.Infof("Finished starting backend.")
- daemon.SdNotify(false, "READY=1")
- // Kube subnet mgr doesn't lease the subnet for this node - it just uses the podCidr that's already assigned.
- if opts.kubeSubnetMgr {
- // Wait for the shutdown to be signalled
- <-ctx.Done()
- } else {
- // Block waiting to renew the lease
- _ = MonitorLease(ctx, sm, bn)
- }
- // To get to here, the Cancel signal must have been received or the lease has been revoked.
- exit()
- }
- func exit() {
- // Wait just a second for the cancel signal to propagate everywhere, then just exit cleanly.
- log.Info("Waiting for cancel to propagate...")
- time.Sleep(time.Second)
- log.Info("Exiting...")
- os.Exit(0)
- }
- func shutdown(sigs chan os.Signal, cancel context.CancelFunc) {
- // Wait for the shutdown signal.
- <-sigs
- // Unregister to get default OS nuke behaviour in case we don't exit cleanly
- signal.Stop(sigs)
- log.Info("Starting shutdown...")
- // Call cancel on the context to close everything down.
- cancel()
- log.Info("Sent cancel signal...")
- }
- func getConfig(ctx context.Context, sm subnet.Manager) (*subnet.Config, error) {
- // Retry every second until it succeeds
- for {
- config, err := sm.GetNetworkConfig(ctx)
- if err != nil {
- log.Errorf("Couldn't fetch network config: %s", err)
- } else if config == nil {
- log.Warningf("Couldn't find network config: %s", err)
- } else {
- log.Infof("Found network config - Backend type: %s", config.BackendType)
- return config, nil
- }
- select {
- case <-ctx.Done():
- return nil, errCanceled
- case <-time.After(1 * time.Second):
- fmt.Println("timed out")
- }
- }
- }
- func MonitorLease(ctx context.Context, sm subnet.Manager, bn backend.Network) error {
- // Use the subnet manager to start watching leases.
- evts := make(chan subnet.Event)
- go subnet.WatchLease(ctx, sm, bn.Lease().Subnet, evts)
- renewMargin := time.Duration(opts.subnetLeaseRenewMargin) * time.Minute
- dur := bn.Lease().Expiration.Sub(time.Now()) - renewMargin
- for {
- select {
- case <-time.After(dur):
- err := sm.RenewLease(ctx, bn.Lease())
- if err != nil {
- log.Error("Error renewing lease (trying again in 1 min): ", err)
- dur = time.Minute
- continue
- }
- log.Info("Lease renewed, new expiration: ", bn.Lease().Expiration)
- dur = bn.Lease().Expiration.Sub(time.Now()) - renewMargin
- case e := <-evts:
- switch e.Type {
- case subnet.EventAdded:
- bn.Lease().Expiration = e.Lease.Expiration
- dur = bn.Lease().Expiration.Sub(time.Now()) - renewMargin
- log.Infof("Waiting for %s to renew lease", dur)
- case subnet.EventRemoved:
- log.Error("Lease has been revoked. Shutting down daemon.")
- return errInterrupted
- }
- case <-ctx.Done():
- log.Infof("Stopped monitoring lease")
- return errCanceled
- }
- }
- }
- func LookupExtIface(ifname string, ifregex string) (*backend.ExternalInterface, error) {
- var iface *net.Interface
- var ifaceAddr net.IP
- var err error
- if len(ifname) > 0 {
- if ifaceAddr = net.ParseIP(ifname); ifaceAddr != nil {
- log.Infof("Searching for interface using %s", ifaceAddr)
- iface, err = ip.GetInterfaceByIP(ifaceAddr)
- if err != nil {
- return nil, fmt.Errorf("error looking up interface %s: %s", ifname, err)
- }
- } else {
- iface, err = net.InterfaceByName(ifname)
- if err != nil {
- return nil, fmt.Errorf("error looking up interface %s: %s", ifname, err)
- }
- }
- } else if len(ifregex) > 0 {
- // Use the regex if specified and the iface option for matching a specific ip or name is not used
- ifaces, err := net.Interfaces()
- if err != nil {
- return nil, fmt.Errorf("error listing all interfaces: %s", err)
- }
- // Check IP
- for _, ifaceToMatch := range ifaces {
- ifaceIP, err := ip.GetIfaceIP4Addr(&ifaceToMatch)
- if err != nil {
- // Skip if there is no IPv4 address
- continue
- }
- matched, err := regexp.MatchString(ifregex, ifaceIP.String())
- if err != nil {
- return nil, fmt.Errorf("regex error matching pattern %s to %s", ifregex, ifaceIP.String())
- }
- if matched {
- ifaceAddr = ifaceIP
- iface = &ifaceToMatch
- break
- }
- }
- // Check Name
- if iface == nil && ifaceAddr == nil {
- for _, ifaceToMatch := range ifaces {
- matched, err := regexp.MatchString(ifregex, ifaceToMatch.Name)
- if err != nil {
- return nil, fmt.Errorf("regex error matching pattern %s to %s", ifregex, ifaceToMatch.Name)
- }
- if matched {
- iface = &ifaceToMatch
- break
- }
- }
- }
- // Check that nothing was matched
- if iface == nil {
- return nil, fmt.Errorf("Could not match pattern %s to any of the available network interfaces", ifregex)
- }
- } else {
- log.Info("Determining IP address of default interface")
- if iface, err = ip.GetDefaultGatewayIface(); err != nil {
- return nil, fmt.Errorf("failed to get default interface: %s", err)
- }
- }
- if ifaceAddr == nil {
- ifaceAddr, err = ip.GetIfaceIP4Addr(iface)
- if err != nil {
- return nil, fmt.Errorf("failed to find IPv4 address for interface %s", iface.Name)
- }
- }
- log.Infof("Using interface with name %s and address %s", iface.Name, ifaceAddr)
- if iface.MTU == 0 {
- return nil, fmt.Errorf("failed to determine MTU for %s interface", ifaceAddr)
- }
- var extAddr net.IP
- if len(opts.publicIP) > 0 {
- extAddr = net.ParseIP(opts.publicIP)
- if extAddr == nil {
- return nil, fmt.Errorf("invalid public IP address: %s", opts.publicIP)
- }
- log.Infof("Using %s as external address", extAddr)
- }
- if extAddr == nil {
- log.Infof("Defaulting external address to interface address (%s)", ifaceAddr)
- extAddr = ifaceAddr
- }
- return &backend.ExternalInterface{
- Iface: iface,
- IfaceAddr: ifaceAddr,
- ExtAddr: extAddr,
- }, nil
- }
- func WriteSubnetFile(path string, nw ip.IP4Net, ipMasq bool, bn backend.Network) error {
- dir, name := filepath.Split(path)
- os.MkdirAll(dir, 0755)
- tempFile := filepath.Join(dir, "."+name)
- f, err := os.Create(tempFile)
- if err != nil {
- return err
- }
- // Write out the first usable IP by incrementing
- // sn.IP by one
- sn := bn.Lease().Subnet
- sn.IP += 1
- fmt.Fprintf(f, "FLANNEL_NETWORK=%s\n", nw)
- fmt.Fprintf(f, "FLANNEL_SUBNET=%s\n", sn)
- fmt.Fprintf(f, "FLANNEL_MTU=%d\n", bn.MTU())
- _, err = fmt.Fprintf(f, "FLANNEL_IPMASQ=%v\n", ipMasq)
- f.Close()
- if err != nil {
- return err
- }
- // rename(2) the temporary file to the desired location so that it becomes
- // atomically visible with the contents
- return os.Rename(tempFile, path)
- //TODO - is this safe? What if it's not on the same FS?
- }
- func mustRunHealthz() {
- address := net.JoinHostPort(opts.healthzIP, strconv.Itoa(opts.healthzPort))
- log.Infof("Start healthz server on %s", address)
- http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
- w.WriteHeader(http.StatusOK)
- w.Write([]byte("flanneld is running"))
- })
- if err := http.ListenAndServe(address, nil); err != nil {
- log.Errorf("Start healthz server error. %v", err)
- panic(err)
- }
- }
- func ReadSubnetFromSubnetFile(path string) ip.IP4Net {
- var prevSubnet ip.IP4Net
- if _, err := os.Stat(path); !os.IsNotExist(err) {
- prevSubnetVals, err := godotenv.Read(path)
- if err != nil {
- log.Errorf("Couldn't fetch previous subnet from subnet file at %s: %s", path, err)
- } else if prevSubnetString, ok := prevSubnetVals["FLANNEL_SUBNET"]; ok {
- err = prevSubnet.UnmarshalJSON([]byte(prevSubnetString))
- if err != nil {
- log.Errorf("Couldn't parse previous subnet from subnet file at %s: %s", path, err)
- }
- }
- }
- return prevSubnet
- }
|