// Starts background goroutines. If file, manifest_url, or address are empty,
// they are not watched. Never returns.
func (kl *Kubelet) RunKubelet(file, manifest_url, etcd_servers, address string, port uint) {
	fileChannel := make(chan api.ContainerManifest)
	etcdChannel := make(chan []api.ContainerManifest)
	httpChannel := make(chan api.ContainerManifest)
	serverChannel := make(chan api.ContainerManifest)

	go util.Forever(func() { kl.WatchFile(file, fileChannel) }, 20*time.Second)
	if manifest_url != "" {
		go util.Forever(func() { kl.WatchHTTP(manifest_url, httpChannel) }, 20*time.Second)
	}
	if etcd_servers != "" {
		servers := []string{etcd_servers}
		log.Printf("Creating etcd client pointing to %v", servers)
		kl.Client = etcd.NewClient(servers)
		go util.Forever(func() { kl.SyncAndSetupEtcdWatch(etcdChannel) }, 20*time.Second)
	}
	if address != "" {
		log.Printf("Starting to listen on %s:%d", address, port)
		handler := KubeletServer{
			Kubelet:       kl,
			UpdateChannel: serverChannel,
		}
		s := &http.Server{
			// TODO: This is broken if address is an ipv6 address.
			Addr:           fmt.Sprintf("%s:%d", address, port),
			Handler:        &handler,
			ReadTimeout:    10 * time.Second,
			WriteTimeout:   10 * time.Second,
			MaxHeaderBytes: 1 << 20,
		}
		go util.Forever(func() { s.ListenAndServe() }, 0)
	}
	kl.RunSyncLoop(etcdChannel, fileChannel, serverChannel, httpChannel, kl)
}

func (c *config) runKubelet() {
	rootDirectory := path.Clean("/var/lib/openshift")
	minionHost := c.bindAddr
	minionPort := 10250

	cadvisorClient, err := cadvisor.NewClient("http://" + c.masterHost + ":4194")
	if err != nil {
		glog.Errorf("Error on creating cadvisor client: %v", err)
	}

	dockerClient, dockerAddr := c.Docker.GetClientOrExit()
	if err := dockerClient.Ping(); err != nil {
		glog.Errorf("WARNING: Docker could not be reached at %s.  Docker must be installed and running to start containers.\n%v", dockerAddr, err)
	} else {
		glog.Infof("Connecting to Docker at %s", dockerAddr)
	}

	etcdClient, _ := c.getEtcdClient()

	// initialize Kubelet
	os.MkdirAll(rootDirectory, 0750)
	cfg := kconfig.NewPodConfig(kconfig.PodConfigNotificationSnapshotAndUpdates)
	kconfig.NewSourceEtcd(kconfig.EtcdKeyForHost(minionHost), etcdClient, cfg.Channel("etcd"))
	k := kubelet.NewMainKubelet(
		minionHost,
		dockerClient,
		cadvisorClient,
		etcdClient,
		rootDirectory,
		30*time.Second)
	go util.Forever(func() { k.Run(cfg.Updates()) }, 0)
	go util.Forever(func() {
		kubelet.ListenAndServeKubeletServer(k, cfg.Channel("http"), minionHost, uint(minionPort))
	}, 0)
}

func startComponents(manifestURL string) (apiServerURL string) {
	// Setup
	servers := []string{"http://localhost:4001"}
	glog.Infof("Creating etcd client pointing to %v", servers)
	machineList := []string{"localhost", "machine"}

	handler := delegateHandler{}
	apiServer := httptest.NewServer(&handler)

	etcdClient := etcd.NewClient(servers)

	cl := client.New(apiServer.URL, nil)
	cl.PollPeriod = time.Second * 1
	cl.Sync = true

	// Master
	m := master.New(&master.Config{
		Client:        cl,
		EtcdServers:   servers,
		Minions:       machineList,
		PodInfoGetter: fakePodInfoGetter{},
	})
	storage, codec := m.API_v1beta1()
	handler.delegate = apiserver.Handle(storage, codec, "/api/v1beta1")

	// Scheduler
	scheduler.New((&factory.ConfigFactory{cl}).Create()).Run()

	controllerManager := controller.NewReplicationManager(cl)

	// Prove that controllerManager's watch works by making it not sync until after this
	// test is over. (Hopefully we don't take 10 minutes!)
	controllerManager.Run(10 * time.Minute)

	// Kubelet (localhost)
	cfg1 := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
	config.NewSourceEtcd(config.EtcdKeyForHost(machineList[0]), etcdClient, cfg1.Channel("etcd"))
	config.NewSourceURL(manifestURL, 5*time.Second, cfg1.Channel("url"))
	myKubelet := kubelet.NewIntegrationTestKubelet(machineList[0], &fakeDocker1)
	go util.Forever(func() { myKubelet.Run(cfg1.Updates()) }, 0)
	go util.Forever(func() {
		kubelet.ListenAndServeKubeletServer(myKubelet, cfg1.Channel("http"), "localhost", 10250)
	}, 0)

	// Kubelet (machine)
	// Create a second kubelet so that the guestbook example's two redis slaves both
	// have a place they can schedule.
	cfg2 := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
	config.NewSourceEtcd(config.EtcdKeyForHost(machineList[1]), etcdClient, cfg2.Channel("etcd"))
	otherKubelet := kubelet.NewIntegrationTestKubelet(machineList[1], &fakeDocker2)
	go util.Forever(func() { otherKubelet.Run(cfg2.Updates()) }, 0)
	go util.Forever(func() {
		kubelet.ListenAndServeKubeletServer(otherKubelet, cfg2.Channel("http"), "localhost", 10251)
	}, 0)

	return apiServer.URL
}

// Starts up a controller manager. Never returns.
func controller_manager() {
	controllerManager := registry.MakeReplicationManager(etcd.NewClient([]string{*etcd_server}),
		client.Client{
			Host: fmt.Sprintf("http://%s:%d", *master_address, *master_port),
		})

	go util.Forever(func() { controllerManager.Synchronize() }, 20*time.Second)
	go util.Forever(func() { controllerManager.WatchControllers() }, 20*time.Second)
	select {}
}

func startKubelet(k *kubelet.Kubelet, cfg *config.PodConfig, kc *KubeletConfig) {
	// start the kubelet
	go util.Forever(func() { k.Run(cfg.Updates()) }, 0)

	// start the kubelet server
	if kc.EnableServer {
		go util.Forever(func() {
			kubelet.ListenAndServeKubeletServer(k, net.IP(kc.Address), kc.Port, kc.EnableDebuggingHandlers)
		}, 0)
	}
}

// Run starts an asynchronous loop that monitors the status of cluster nodes.
func (nc *NodeController) Run(period time.Duration) {
	// Incorporate the results of node status pushed from kubelet to master.
	go util.Forever(func() {
		if err := nc.monitorNodeStatus(); err != nil {
			glog.Errorf("Error monitoring node status: %v", err)
		}
	}, nc.nodeMonitorPeriod)

	go util.Forever(func() {
		nc.podEvictor.TryEvict(func(nodeName string) { nc.deletePods(nodeName) })
	}, nodeEvictionPeriod)
}

// Run creates initial node list and start syncing instances from cloudprovider if any.
// It also starts syncing cluster node status.
// 1. RegisterNodes() is called only once to register all initial nodes (from cloudprovider
//    or from command line flag). To make cluster bootstrap faster, node controller populates
//    node addresses.
// 2. SyncCloud() is called periodically (if enabled) to sync instances from cloudprovider.
//    Node created here will only have specs.
// 3. SyncNodeStatus() is called periodically (if enabled) to sync node status for nodes in
//    k8s cluster.
func (s *NodeController) Run(period time.Duration, syncNodeList, syncNodeStatus bool) {
	// Register intial set of nodes with their status set.
	var nodes *api.NodeList
	var err error
	if s.isRunningCloudProvider() {
		if syncNodeList {
			nodes, err = s.GetCloudNodesWithSpec()
			if err != nil {
				glog.Errorf("Error loading initial node from cloudprovider: %v", err)
			}
		} else {
			nodes = &api.NodeList{}
		}
	} else {
		nodes, err = s.GetStaticNodesWithSpec()
		if err != nil {
			glog.Errorf("Error loading initial static nodes: %v", err)
		}
	}
	nodes, err = s.PopulateAddresses(nodes)
	if err != nil {
		glog.Errorf("Error getting nodes ips: %v", err)
	}
	if err = s.RegisterNodes(nodes, s.registerRetryCount, period); err != nil {
		glog.Errorf("Error registering node list %+v: %v", nodes, err)
	}

	// Start syncing node list from cloudprovider.
	if syncNodeList && s.isRunningCloudProvider() {
		go util.Forever(func() {
			if err = s.SyncCloud(); err != nil {
				glog.Errorf("Error syncing cloud: %v", err)
			}
		}, period)
	}

	if syncNodeStatus {
		// Start syncing node status.
		go util.Forever(func() {
			if err = s.SyncNodeStatus(); err != nil {
				glog.Errorf("Error syncing status: %v", err)
			}
		}, period)
	} else {
		// Start checking node reachability and evicting timeouted pods.
		go util.Forever(func() {
			if err = s.EvictTimeoutedPods(); err != nil {
				glog.Errorf("Error evicting timeouted pods: %v", err)
			}
		}, period)
	}
}

// RunKubelet starts background goroutines. If config_path, manifest_url, or address are empty,
// they are not watched. Never returns.
func (kl *Kubelet) RunKubelet(dockerEndpoint, configPath, manifestURL, etcdServers, address string, port uint) {
	if kl.CadvisorClient == nil {
		var err error
		kl.CadvisorClient, err = cadvisor.NewClient("http://127.0.0.1:5000")
		if err != nil {
			glog.Errorf("Error on creating cadvisor client: %v", err)
		}
	}
	if kl.DockerPuller == nil {
		kl.DockerPuller = NewDockerPuller(kl.DockerClient)
	}
	updateChannel := make(chan manifestUpdate)
	if configPath != "" {
		glog.Infof("Watching for file configs at %s", configPath)
		go util.Forever(func() {
			kl.WatchFiles(configPath, updateChannel)
		}, kl.FileCheckFrequency)
	}
	if manifestURL != "" {
		glog.Infof("Watching for HTTP configs at %s", manifestURL)
		go util.Forever(func() {
			if err := kl.extractFromHTTP(manifestURL, updateChannel); err != nil {
				glog.Errorf("Error syncing http: %v", err)
			}
		}, kl.HTTPCheckFrequency)
	}
	if etcdServers != "" {
		servers := []string{etcdServers}
		glog.Infof("Watching for etcd configs at %v", servers)
		kl.EtcdClient = etcd.NewClient(servers)
		go util.Forever(func() { kl.SyncAndSetupEtcdWatch(updateChannel) }, 20*time.Second)
	}
	if address != "" {
		glog.Infof("Starting to listen on %s:%d", address, port)
		handler := Server{
			Kubelet:         kl,
			UpdateChannel:   updateChannel,
			DelegateHandler: http.DefaultServeMux,
		}
		s := &http.Server{
			Addr:           net.JoinHostPort(address, strconv.FormatUint(uint64(port), 10)),
			Handler:        &handler,
			ReadTimeout:    10 * time.Second,
			WriteTimeout:   10 * time.Second,
			MaxHeaderBytes: 1 << 20,
		}
		go util.Forever(func() { s.ListenAndServe() }, 0)
	}
	kl.HealthChecker = health.MakeHealthChecker()
	kl.syncLoop(updateChannel, kl)
}

func main() {
	client.BindClientConfigFlags(flag.CommandLine, clientConfig)
	flag.Set("logtostderr", "true")
	flag.Parse()

	cmd := exec.Command("haproxy", "-f", configPath, "-p", "/var/run/haproxy.pid")
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	err := cmd.Run()
	if err != nil {
		if o, err := cmd.CombinedOutput(); err != nil {
			glog.Error(string(o))
		}
		glog.Fatalf("haproxy process died, : %v", err)
	}
	glog.Info("started haproxy")

	clientConfig.TLSClientConfig = loadTLSConfigOrDie()
	//clientConfig.Version = "v1beta3"
	clientConfig.Version = "v1"
	kubeClient, err := client.New(clientConfig)
	if err != nil {
		glog.Fatalf("Invalid API configuration: %v", err)
	}

	cu := configUpdater{
		make([]api.Endpoints, 0),
		make([]api.Service, 0),
		make(chan []api.Endpoints),
		make(chan []api.Service),
		template.Must(template.ParseFiles(templatePath)),
		make(chan struct{}, 1),
	}

	endpointsConfig := config.NewEndpointsConfig()
	serviceConfig := config.NewServiceConfig()
	endpointsConfig.RegisterHandler(cu.eu)
	serviceConfig.RegisterHandler(cu.su)

	config.NewSourceAPI(
		kubeClient.Services(api.NamespaceAll),
		kubeClient.Endpoints(api.NamespaceAll),
		30*time.Second,
		serviceConfig.Channel("api"),
		endpointsConfig.Channel("api"),
	)
	glog.Info("started watch")

	go util.Forever(cu.reloadLoop, 1*time.Second)
	util.Forever(cu.syncLoop, 1*time.Second)
}

// Starts api services (the master). Never returns.
func api_server() {
	machineList := util.StringList{*kubelet_address}

	etcdClient := etcd.NewClient([]string{*etcd_server})
	podRegistry := registry.MakeEtcdRegistry(etcdClient, machineList)
	controllerRegistry := registry.MakeEtcdRegistry(etcdClient, machineList)
	serviceRegistry := registry.MakeEtcdRegistry(etcdClient, machineList)

	containerInfo := &client.HTTPContainerInfo{
		Client: http.DefaultClient,
		Port:   *kubelet_port,
	}
	random := rand.New(rand.NewSource(int64(time.Now().Nanosecond())))

	storage := map[string]apiserver.RESTStorage{
		"pods": registry.MakePodRegistryStorage(podRegistry, containerInfo, registry.MakeFirstFitScheduler(machineList, podRegistry, random)),
		"replicationControllers": registry.MakeControllerRegistryStorage(controllerRegistry),
		"services":               registry.MakeServiceRegistryStorage(serviceRegistry),
	}

	endpoints := registry.MakeEndpointController(serviceRegistry, podRegistry)
	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)

	s := &http.Server{
		Addr:           fmt.Sprintf("%s:%d", *master_address, *master_port),
		Handler:        apiserver.New(storage, *apiPrefix),
		ReadTimeout:    10 * time.Second,
		WriteTimeout:   10 * time.Second,
		MaxHeaderBytes: 1 << 20,
	}
	log.Fatal(s.ListenAndServe())
}

// Sync with etcd, and set up an etcd watch for new configurations
// The channel to send new configurations across
// This function loops forever and is intended to be run in a go routine.
func (kl *Kubelet) SyncAndSetupEtcdWatch(changeChannel chan<- []api.ContainerManifest) {
	hostname, err := exec.Command("hostname", "-f").Output()
	if err != nil {
		log.Printf("Couldn't determine hostname : %v", err)
		return
	}
	key := "/registry/hosts/" + strings.TrimSpace(string(hostname))
	// First fetch the initial configuration (watch only gives changes...)
	for {
		err = kl.getKubeletStateFromEtcd(key, changeChannel)
		if err == nil {
			// We got a successful response, etcd is up, set up the watch.
			break
		}
		time.Sleep(30 * time.Second)
	}

	done := make(chan bool)
	go util.Forever(func() { kl.TimeoutWatch(done) }, 0)
	for {
		// The etcd client will close the watch channel when it exits.  So we need
		// to create and service a new one every time.
		watchChannel := make(chan *etcd.Response)
		// We don't push this through Forever because if it dies, we just do it again in 30 secs.
		// anyway.
		go kl.WatchEtcd(watchChannel, changeChannel)

		kl.getKubeletStateFromEtcd(key, changeChannel)
		log.Printf("Setting up a watch for configuration changes in etcd for %s", key)
		kl.Client.Watch(key, 0, true, watchChannel, done)
	}
}

// NewOperations returns a new Operations repository.
func NewOperations() *Operations {
	ops := &Operations{
		ops: map[string]*Operation{},
	}
	go util.Forever(func() { ops.expire(10 * time.Minute) }, 5*time.Minute)
	return ops
}

// RunControllerManager starts a controller
func RunControllerManager(machineList []string, cl *client.Client, nodeMilliCPU, nodeMemory int64) {
	if int64(int(nodeMilliCPU)) != nodeMilliCPU {
		glog.Warningf("node_milli_cpu is too big for platform. Clamping: %d -> %d",
			nodeMilliCPU, math.MaxInt32)
		nodeMilliCPU = math.MaxInt32
	}

	if int64(int(nodeMemory)) != nodeMemory {
		glog.Warningf("node_memory is too big for platform. Clamping: %d -> %d",
			nodeMemory, math.MaxInt32)
		nodeMemory = math.MaxInt32
	}

	nodeResources := &api.NodeResources{
		Capacity: api.ResourceList{
			resources.CPU:    util.NewIntOrStringFromInt(int(nodeMilliCPU)),
			resources.Memory: util.NewIntOrStringFromInt(int(nodeMemory)),
		},
	}
	minionController := minionControllerPkg.NewMinionController(nil, "", machineList, nodeResources, cl)
	minionController.Run(10 * time.Second)

	endpoints := service.NewEndpointController(cl)
	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)

	controllerManager := controller.NewReplicationManager(cl)
	controllerManager.Run(10 * time.Second)
}

func (m *Master) init(cloud cloudprovider.Interface, podInfoGetter client.PodInfoGetter) {
	podCache := NewPodCache(podInfoGetter, m.podRegistry, time.Second*30)
	go podCache.Loop()

	endpoints := endpoint.NewEndpointController(m.serviceRegistry, m.client)
	go util.Forever(func() { endpoints.SyncServiceEndpoints() }, time.Second*10)

	random := rand.New(rand.NewSource(int64(time.Now().Nanosecond())))
	s := scheduler.NewRandomFitScheduler(m.podRegistry, random)
	m.storage = map[string]apiserver.RESTStorage{
		"pods": pod.NewRegistryStorage(&pod.RegistryStorageConfig{
			CloudProvider: cloud,
			MinionLister:  m.minionRegistry,
			PodCache:      podCache,
			PodInfoGetter: podInfoGetter,
			Registry:      m.podRegistry,
			Scheduler:     s,
		}),
		"replicationControllers": controller.NewRegistryStorage(m.controllerRegistry, m.podRegistry),
		"services":               service.NewRegistryStorage(m.serviceRegistry, cloud, m.minionRegistry),
		"minions":                minion.NewRegistryStorage(m.minionRegistry),

		// TODO: should appear only in scheduler API group.
		"bindings": binding.NewBindingStorage(m.bindingRegistry),
	}
}

func startKubelet(k KubeletBootstrap, podCfg *config.PodConfig, kc *KubeletConfig) {
	// start the kubelet
	go util.Forever(func() { k.Run(podCfg.Updates()) }, 0)

	// start the kubelet server
	if kc.EnableServer {
		go util.Forever(func() {
			k.ListenAndServe(net.IP(kc.Address), kc.Port, kc.TLSOptions, kc.EnableDebuggingHandlers)
		}, 0)
	}
	if kc.ReadOnlyPort > 0 {
		go util.Forever(func() {
			k.ListenAndServeReadOnly(net.IP(kc.Address), kc.ReadOnlyPort)
		}, 0)
	}
}

// Sync with etcd, and set up an etcd watch for new configurations
// The channel to send new configurations across
// This function loops forever and is intended to be run in a go routine.
func (kl *Kubelet) SyncAndSetupEtcdWatch(updateChannel chan<- manifestUpdate) {
	key := "/registry/hosts/" + strings.TrimSpace(kl.Hostname)
	// First fetch the initial configuration (watch only gives changes...)
	for {
		err := kl.getKubeletStateFromEtcd(key, updateChannel)
		if err == nil {
			// We got a successful response, etcd is up, set up the watch.
			break
		}
		time.Sleep(30 * time.Second)
	}

	done := make(chan bool)
	go util.Forever(func() { kl.TimeoutWatch(done) }, 0)
	for {
		// The etcd client will close the watch channel when it exits.  So we need
		// to create and service a new one every time.
		watchChannel := make(chan *etcd.Response)
		// We don't push this through Forever because if it dies, we just do it again in 30 secs.
		// anyway.
		go kl.WatchEtcd(watchChannel, updateChannel)

		kl.getKubeletStateFromEtcd(key, updateChannel)
		glog.V(1).Infof("Setting up a watch for configuration changes in etcd for %s", key)
		kl.EtcdClient.Watch(key, 0, true, watchChannel, done)
	}
}

// RunKubelet starts a Kubelet talking to dockerEndpoint
func RunKubelet(etcdClient tools.EtcdClient, hostname, dockerEndpoint string) {
	dockerClient, err := docker.NewClient(GetDockerEndpoint(dockerEndpoint))
	if err != nil {
		glog.Fatal("Couldn't connect to docker.")
	}

	// Kubelet (localhost)
	os.MkdirAll(testRootDir, 0750)
	cfg1 := config.NewPodConfig(config.PodConfigNotificationSnapshotAndUpdates)
	config.NewSourceEtcd(config.EtcdKeyForHost(hostname), etcdClient, cfg1.Channel("etcd"))
	myKubelet := kubelet.NewIntegrationTestKubelet(hostname, testRootDir, dockerClient)
	go util.Forever(func() { myKubelet.Run(cfg1.Updates()) }, 0)
	go util.Forever(func() {
		kubelet.ListenAndServeKubeletServer(myKubelet, cfg1.Channel("http"), net.ParseIP("127.0.0.1"), 10250, true)
	}, 0)
}

// Run starts syncing instances from cloudprovider periodically, or create initial minion list.
func (s *MinionController) Run(period time.Duration) {
	if s.cloud != nil && len(s.matchRE) > 0 {
		go util.Forever(func() { s.SyncCloud() }, period)
	} else {
		go s.SyncStatic(period)
	}
}

func (rc *RouteController) Run(syncPeriod time.Duration) {
	go util.Forever(func() {
		if err := rc.reconcileNodeRoutes(); err != nil {
			glog.Errorf("Couldn't reconcile node routes: %v", err)
		}
	}, syncPeriod)
}

func (kl *Kubelet) GarbageCollectLoop() {
	util.Forever(func() {
		if err := kl.GarbageCollectContainers(); err != nil {
			glog.Errorf("Garbage collect failed: %v", err)
		}
	}, time.Minute*1)
}