Add --disable_watches flag

Setting this flag to true makes fleet not set any watches in etcd. This
should aid in making fleet scale better.

Documentation/fleet-scaling.md

@@ -55,3 +55,9 @@ wins:
 * Making some defaults exported and allow them to be overridden. For instance
     fleet's tokenLimit controls how many Units are listed per "page". *See the
     `--token-limit` flag.*
+
+* Removing watches from fleet: By removing the watches from fleet we stop
+    the entire cluster from walking up whenever a new job is to be scheduled.
+    The downside of this change is that fleet's responsiveness is lower.
+    *See the `--disable-watches` flag.*
+

config/config.go

@@ -32,6 +32,7 @@ type Config struct {
 	AgentTTL                string
 	TokenLimit              int
 	DisableEngine           bool
+	DisableWatches          bool
 	VerifyUnits             bool
 	AuthorizedKeysFile      string
 }

fleetd/fleetd.go

@@ -82,6 +82,7 @@ func main() {
 	cfgset.String("agent_ttl", agent.DefaultTTL, "TTL in seconds of fleet machine state in etcd")
 	cfgset.Int("token_limit", 100, "Maximum number of entries per page returned from API requests")
 	cfgset.Bool("disable_engine", false, "Disable the engine entirely, use with care")
+	cfgset.Bool("disable_watches", false, "Disable the use of etcd watches. Increases scheduling latency")
 	cfgset.Bool("verify_units", false, "DEPRECATED - This option is ignored")
 	cfgset.String("authorized_keys_file", "", "DEPRECATED - This option is ignored")
 
@@ -195,6 +196,7 @@ func getConfig(flagset *flag.FlagSet, userCfgFile string) (*config.Config, error
 		RawMetadata:             (*flagset.Lookup("metadata")).Value.(flag.Getter).Get().(string),
 		AgentTTL:                (*flagset.Lookup("agent_ttl")).Value.(flag.Getter).Get().(string),
 		DisableEngine:           (*flagset.Lookup("disable_engine")).Value.(flag.Getter).Get().(bool),
+		DisableWatches:          (*flagset.Lookup("disable_watches")).Value.(flag.Getter).Get().(bool),
 		VerifyUnits:             (*flagset.Lookup("verify_units")).Value.(flag.Getter).Get().(bool),
 		TokenLimit:              (*flagset.Lookup("token_limit")).Value.(flag.Getter).Get().(int),
 		AuthorizedKeysFile:      (*flagset.Lookup("authorized_keys_file")).Value.(flag.Getter).Get().(string),

server/server.go

@@ -100,7 +100,10 @@ func New(cfg config.Config) (*Server, error) {
 
 	a := agent.New(mgr, gen, reg, mach, agentTTL)
 
-	rStream := registry.NewEtcdEventStream(kAPI, cfg.EtcdKeyPrefix)
+	var rStream pkg.EventStream
+	if !cfg.DisableWatches {
+		rStream = registry.NewEtcdEventStream(kAPI, cfg.EtcdKeyPrefix)
+	}
 	lManager := lease.NewEtcdLeaseManager(kAPI, cfg.EtcdKeyPrefix, etcdRequestTimeout)
 
 	ar := agent.NewReconciler(reg, rStream)