func getHosts() []string {
	port := config.AtPath("hailo", "service", "cassandra", "defaults", "cqlPort").AsInt(defaultPort)
	hosts := config.AtPath("hailo", "service", "cassandra", hostsCfgKey()).AsHostnameArray(port)
	if len(hosts) > 0 {
		return hosts
	}

	// No hosts returned: try DNS
	tier := config.AtPath("hailo", "service", "cassandra", "tier").AsString("premium")
	hosts, err := dns.Hosts("cassandra-" + tier)
	if err != nil {
		log.Errorf("[Cassandra] Failed to load hosts from DNS: %s", err.Error())
		return defaultHosts
	}

	if len(hosts) == 0 {
		return defaultHosts
	}
	// We need to append the port to hosts coming from DNS
	for i, host := range hosts {
		hosts[i] = host + fmt.Sprintf(":%d", port)
	}

	return hosts
}

func getHosts(port int, path ...string) []string {
	if hosts := config.AtPath(path...).AsHostnameArray(port); len(hosts) > 0 {
		return hosts
	}

	// should we lookup dns?
	if config.AtPath("hailo", "service", "nsq", "disableDnsLookup").AsBool() {
		return []string{}
	}

	// try dns lookup
	cluster := config.AtPath("hailo", "service", "nsq", "cluster").AsString("general")
	hosts, err := dns.Hosts("nsq-" + cluster)
	if err != nil {
		log.Errorf("Failed to load NSQ hosts from dns: %v", err)
		return []string{}
	}

	// append port
	for i, host := range hosts {
		hosts[i] = fmt.Sprintf("%s:%d", host, port)
	}

	return hosts
}

func getHosts() []string {
	cassandraHostKey := getCassandraHostConfigKey()
	config.WaitUntilLoaded(5 * time.Second)
	port := config.AtPath("hailo", "service", "cassandra", "defaults", "thriftPort").AsInt(defaultPort)
	if hosts := config.AtPath("hailo", "service", "cassandra", cassandraHostKey).AsHostnameArray(port); len(hosts) > 0 {
		return hosts
	}

	// No hosts returned: try DNS
	tier := config.AtPath("hailo", "service", "cassandra", "tier").AsString("premium")
	hosts, err := dns.Hosts("cassandra-" + tier)
	if err != nil {
		log.Errorf("Failed to load Cassandra hosts from dns: %v", err)
		return defaultHosts
	}

	if len(hosts) == 0 {
		return defaultHosts
	}
	// We need to append the port to hosts coming from DNS
	for i, host := range hosts {
		hosts[i] = host + fmt.Sprintf(":%d", port)
	}

	return hosts
}

func getHosts() []string {
	hostsConfigPath := []string{"hailo", "service", "zookeeper", "hosts"}
	tier := config.AtPath("hailo", "service", "zookeeper", "tier").AsString("general")
	if tier != "general" {
		hostsConfigPath = append(hostsConfigPath, tier)
	}

	if hosts := config.AtPath(hostsConfigPath...).AsHostnameArray(2181); len(hosts) > 0 {
		return hosts
	}

	// no hosts returned so try dns
	hosts, err := dns.Hosts("zookeeper-" + tier)
	if err != nil {
		log.Errorf("Failed to load ZK hosts from dns: %v", err)
		return []string{"localhost:2181"}
	}

	// for safety fall back to localhost
	if len(hosts) == 0 {
		return []string{"localhost:2181"}
	}

	// append port
	for i, host := range hosts {
		hosts[i] = host + ":2181"
	}

	return hosts
}

func getHosts() []string {
	hostConfigPath := []string{"hailo", "service", "memcache", "servers"}
	host := "memcached"

	// check if tier is specified and act accordingly
	tier := config.AtPath("hailo", "service", "memcache", "tier").AsString("")
	if tier != "" {
		hostConfigPath = append(hostConfigPath, tier)
		host = fmt.Sprintf("%s-%s", host, tier)
	}

	if hosts := config.AtPath(hostConfigPath...).AsHostnameArray(11211); len(hosts) > 0 {
		return hosts
	}

	// no hosts returned so try dns
	hosts, err := dns.Hosts(host)
	if err != nil {
		log.Errorf("[Memcache] Failed to load hosts from dns, returning empty list: %v", err)
		return []string{}
	}

	// append port
	for i, host := range hosts {
		hosts[i] = host + ":11211"
	}

	return hosts
}

func newdefaultClient() MemcacheClient {
	serverSelector := new(memcache.ServerList)
	client := memcache.NewFromSelector(serverSelector)

	// Listen for config changes
	ch := config.SubscribeChanges()
	go func() {
		for _ = range ch {
			loadFromConfig(serverSelector, client)
		}
	}()

	loadFromConfig(serverSelector, client)

	// Log on init
	hosts := config.AtPath("hailo", "service", "memcache", "servers").AsHostnameArray(11211)
	operationTimeout := config.AtPath("hailo", "service", "memcache", "timeouts", "dialTimeout").
		AsDuration(defaultDialTimeout)
	dialTimeout := config.AtPath("hailo", "service", "memcache", "timeouts", "operationTimeout").
		AsDuration(defaultOperationTimeout)

	log.Infof("[Memcache] Initialising Memcache client to hosts %v: dial timeout %v, op timeout: %v", hosts,
		dialTimeout, operationTimeout)

	return client
}

func createCircuit(service, endpoint string) Circuit {
	options := defaultOptions
	config.AtPath("hailo", "platform", "circuitbreaker").AsStruct(&options)
	config.AtPath("hailo", "platform", "circuitbreaker", "endpoints", service, endpoint).AsStruct(&options)

	log.Debugf("Circuitbreaker config for %s.%s: %#v", service, endpoint, options)
	return NewDefaultCircuit(options)
}

func (rs *RedisDedupeClient) connect() (*redis.Pool, error) {
	host := config.AtPath("hailo", "service", "deduper", "redis", "hostname").AsString(":16379")
	// var password string
	log.Debugf("Setting redis server from config: %v", host)
	pool := &redis.Pool{
		MaxIdle:     3,
		IdleTimeout: 240 * time.Second,
		Dial: func() (redis.Conn, error) {
			c, err := redis.Dial("tcp", host)
			if err != nil {
				return nil, err
			}
			if _, err := c.Do("PING"); err != nil {
				c.Close()
				return nil, err
			}
			return c, err
		},
		TestOnBorrow: func(c redis.Conn, t time.Time) error {
			_, err := c.Do("PING")
			return err
		},
	}

	return pool, nil
}

func (s *DefaultSubscriber) AddHandlers(handler nsqlib.Handler) {
	subHandlers := config.AtPath("hailo", "service", "nsq", "subHandlers").AsInt(6)
	log.Infof("Adding %d handlers", subHandlers)
	for i := 0; i < subHandlers; i++ {
		s.AddHandler(handler)
	}
}

func loadFromConfig(sl *memcache.ServerList, client *memcache.Client) {
	hosts := getHosts()
	log.Tracef("[Memcache] Setting memcache servers from config: %v", hosts)
	err := sl.SetServers(hosts...)
	if err != nil {
		log.Errorf("[Memcache] Error setting memcache servers: %v", err)
	}

	// Technically we have a race here since the timeouts are not protected by a mutex, however it isn't really a
	// problem if the timeout is stale for a short period.
	client.Timeout = config.AtPath("hailo", "service", "memcache", "timeouts", "operationTimeout").
		AsDuration(defaultOperationTimeout)
	log.Tracef("[Memcache] Set Memcache operation timeout from config: %v", client.Timeout)
	client.DialTimeout = config.AtPath("hailo", "service", "memcache", "timeouts", "dialTimeout").
		AsDuration(defaultDialTimeout)
	log.Tracef("[Memcache] Set Memcache dial timeout from config: %v", client.DialTimeout)
}

// loadFromConfig will grab the configurable settings from config service
func (t *Timeout) loadFromConfig() {
	min := config.AtPath("hailo", "platform", "timeout", "min").AsDuration(defaultMin)
	max := config.AtPath("hailo", "platform", "timeout", "max").AsDuration(defaultMax)
	multiplier := config.AtPath("hailo", "platform", "timeout", "multiplier").AsFloat64(defaultMultiplier)

	// any difference?
	if hashTimeouts(min, max, multiplier) == t.hashTimeouts() {
		return
	}

	t.Lock()
	defer t.Unlock()
	t.min = min
	t.max = max
	t.multiplier = multiplier

	log.Infof("[Client] Loaded timeout configuration from config service [min=%v, max=%v, multiplier=%v]", min, max, multiplier)
}

// NewGlobalLocker returns a global leader which is basically just a region leader pinned to one region based on
// config.
func NewGlobalLeader(id string) Leader {
	for {
		if config.AtPath("leaders", "isLeader").AsBool() {
			break
		}
		<-config.SubscribeChanges()
	}
	return RegionLeader(id)
}

func loadAccConfig(path ...string) []*AWSAccount {
	bytes := config.AtPath(path...).AsJson()
	accs := make([]*AWSAccount, 0)
	err := json.Unmarshal(bytes, &accs)
	if err != nil {
		log.Warnf("[AWS Manager] Failed to unmarshal AWS credential pairs from config: %s", err)
		return nil
	}
	return accs
}

func TestPub(t *testing.T) {
	config.LoadFromService("testservice")
	s := config.AtPath("configService", "hash").AsString("default")
	if s == "default" {
		t.Fatal("Failed to load config from config service")
	}
	err := Publish("testtopic", []byte("This is my payload"))
	if err != nil {
		t.Error(fmt.Sprintf("Failed to PUB: %v", err))
	}
}

func loadEndpointConfig() {
	log.Info("Loading ElasticSearch config")

	port := config.AtPath("hailo", "service", "elasticsearch", "port").AsInt(9200)
	hosts := config.AtPath("hailo", "service", "elasticsearch", "hosts").AsHostnameArray(port)

	if len(hosts) == 0 {
		hosts = append(hosts, "localhost:19200")
	}

	// Set these hosts in the Elasticsearch library
	// This will initialise a host pool which uses an Epsilon Greedy algorithm to find healthy hosts
	// and send to requests to them, and not unhealthy or slow hosts
	eapi.Port = strconv.Itoa(port)
	if port == 443 {
		eapi.Protocol = "https"
	}
	eapi.SetHosts(hosts)

	log.Infof("ElasticSearch hosts loaded: %v", eapi.Hosts)
}

// ksAuth returns the username and password for the given keyspace
func ksAuth(ks string) (string, string, error) {
	if !config.AtPath("hailo", "service", "cassandra", "authentication", "enabled").AsBool() {
		return "", "", nil
	}

	confJson := config.AtPath("hailo", "service", "cassandra", "authentication", "keyspaces").AsJson()
	rawConf := make(map[string]map[string]string, 5)
	if err := json.Unmarshal(confJson, &rawConf); err != nil {
		// Don't log raw data in an attempt to not log our passwords
		log.Warnf("[Cassandra] Failed to unmarshal authentication configuration: %s", err.Error())
		return "", "", err
	}

	for candidateKs, v := range rawConf {
		if candidateKs == ks {
			return v["username"], v["password"], nil
		}
	}

	return "", "", nil
}

func getKsConfig(ks string) (ksConfig, error) {
	if !config.WaitUntilLoaded(5 * time.Second) {
		return ksConfig{}, fmt.Errorf("Config not loaded")
	}

	username, password, err := ksAuth(ks)
	if err != nil {
		return ksConfig{}, err
	}

	c := ksConfig{
		ks:       ks,
		hosts:    getHosts(),
		username: username,
		password: password,
		retries:  config.AtPath("hailo", "service", "cassandra", "defaults", "maxRetries").AsInt(5),
		cl:       clFromString(config.AtPath("hailo", "service", "cassandra", "defaults", "consistencyLevel").AsString("")),
		timeout:  config.AtPath("hailo", "service", "cassandra", "defaults", "recvTimeout").AsDuration("1s"),
	}
	cc := gocql.NewCluster(c.hosts...)
	cc.ProtoVersion = config.AtPath("hailo", "service", "cassandra", "defaults", "protoVersion").AsInt(2)
	cc.Consistency = c.cl
	cc.Compressor = gocql.SnappyCompressor{}
	cc.DiscoverHosts = false
	cc.NumConns = config.AtPath("hailo", "service", "cassandra", "defaults", "maxHostConns").AsInt(2)
	cc.Authenticator = gocql.PasswordAuthenticator{
		Username: c.username,
		Password: c.password,
	}
	cc.Timeout = c.timeout
	cc.Keyspace = c.ks
	cc.RetryPolicy = &gocql.SimpleRetryPolicy{
		NumRetries: c.retries,
	}
	cc.PoolConfig.HostSelectionPolicy = gocql.HostPoolHostPolicy(
		hostpool.NewEpsilonGreedy(c.hosts, 5*time.Minute, &hostpool.LinearEpsilonValueCalculator{}),
	)
	c.cc = cc
	return c, nil
}

func loadStatsd(addr string) g2s.Statter {
	disabled := config.AtPath("hailo", "service", "instrumentation", "statsd", "disabled").AsBool()
	if disabled {
		return g2s.Noop()
	}

	s, err := g2s.Dial("udp", addr)
	if err != nil {
		log.Warnf("Error initialising statsd connection to %v", addr)
		return nil
	}

	return s
}

// hostsCfgKey gets the config key that should be used to load cassandra hosts. This is to support multiple 'tiered'
// Cassandra clusters.
func hostsCfgKey() string {
	// Check what cluster are we supposed to contact, revert to the default config if not specified
	tier := config.AtPath("hailo", "service", "cassandra", "tier").AsString("")
	if tier == "" {
		tier = defaultTier
	}

	switch tier {
	case "general":
		return "hosts"
	default:
		return fmt.Sprintf("%sHosts", tier)
	}
}

// run is our main healthcheck loop
func (r *results) run() {
	for {
		select {
		// Listen for config changes and update the healthcheck when needed
		case <-config.SubscribeChanges():
			// Allow healthcheck parameters to be overridden in config
			config.AtPath("hailo", "platform", "healthcheck", r.hc.Id).AsStruct(r.hc)
		case <-runNow:
			r.collect()
		case <-time.After(r.hc.Interval):
			r.collect()
		}
	}
}