func (e *periodicExporter) scrapeSlaves() {
	e.slaves.Lock()
	urls := make([]string, len(e.slaves.urls))
	copy(urls, e.slaves.urls)
	e.slaves.Unlock()

	urlCount := len(urls)
	log.Debugf("active slaves: %d", urlCount)

	urlChan := make(chan string)
	metricsChan := make(chan prometheus.Metric)
	go e.setMetrics(metricsChan)

	poolSize := concurrentFetch
	if urlCount < concurrentFetch {
		poolSize = urlCount
	}

	log.Debugf("creating fetch pool of size %d", poolSize)

	var wg sync.WaitGroup
	wg.Add(poolSize)
	for i := 0; i < poolSize; i++ {
		go e.fetch(urlChan, metricsChan, &wg)
	}

	for _, url := range urls {
		urlChan <- url
	}
	close(urlChan)

	wg.Wait()
	close(metricsChan)
}

func (c *runitCollector) Update(ch chan<- prometheus.Metric) error {
	services, err := runit.GetServices("/etc/service")
	if err != nil {
		return err
	}

	for _, service := range services {
		status, err := service.Status()
		if err != nil {
			log.Debugf("Couldn't get status for %s: %s, skipping...", service.Name, err)
			continue
		}

		log.Debugf("%s is %d on pid %d for %d seconds", service.Name, status.State, status.Pid, status.Duration)
		c.state.WithLabelValues(service.Name).Set(float64(status.State))
		c.stateDesired.WithLabelValues(service.Name).Set(float64(status.Want))
		c.stateTimestamp.WithLabelValues(service.Name).Set(float64(status.Timestamp.Unix()))
		if status.NormallyUp {
			c.stateNormal.WithLabelValues(service.Name).Set(1)
		} else {
			c.stateNormal.WithLabelValues(service.Name).Set(0)
		}
	}
	c.state.Collect(ch)
	c.stateDesired.Collect(ch)
	c.stateNormal.Collect(ch)
	c.stateTimestamp.Collect(ch)

	return nil
}

func (c *gmondCollector) setMetric(name, cluster string, metric ganglia.Metric) {
	if _, ok := c.metrics[name]; !ok {
		var desc string
		var title string
		for _, element := range metric.ExtraData.ExtraElements {
			switch element.Name {
			case "DESC":
				desc = element.Val
			case "TITLE":
				title = element.Val
			}
			if title != "" && desc != "" {
				break
			}
		}
		log.Debugf("Register %s: %s", name, desc)
		c.metrics[name] = prometheus.NewGaugeVec(
			prometheus.GaugeOpts{
				Namespace: gangliaNamespace,
				Name:      name,
				Help:      desc,
			},
			[]string{"cluster"},
		)
	}
	log.Debugf("Set %s{cluster=%q}: %f", name, cluster, metric.Value)
	c.metrics[name].WithLabelValues(cluster).Set(metric.Value)
}

// Expose filesystem fullness.
func (c *filesystemCollector) GetStats() (stats []filesystemStats, err error) {
	mpds, err := mountPointDetails()
	if err != nil {
		return nil, err
	}
	stats = []filesystemStats{}
	for _, mpd := range mpds {
		if c.ignoredMountPointsPattern.MatchString(mpd.mountPoint) {
			log.Debugf("Ignoring mount point: %s", mpd.mountPoint)
			continue
		}
		buf := new(syscall.Statfs_t)
		err := syscall.Statfs(mpd.mountPoint, buf)
		if err != nil {
			log.Debugf("Statfs on %s returned %s",
				mpd.mountPoint, err)
			continue
		}

		labelValues := []string{mpd.device, mpd.mountPoint, mpd.fsType}
		stats = append(stats, filesystemStats{
			labelValues: labelValues,
			size:        float64(buf.Blocks) * float64(buf.Bsize),
			free:        float64(buf.Bfree) * float64(buf.Bsize),
			avail:       float64(buf.Bavail) * float64(buf.Bsize),
			files:       float64(buf.Files),
			filesFree:   float64(buf.Ffree),
		})
	}
	return stats, nil
}

// StopScraper implements Target.
func (t *Target) StopScraper() {
	log.Debugf("Stopping scraper for target %v...", t)

	close(t.scraperStopping)
	<-t.scraperStopped

	log.Debugf("Scraper for target %v stopped.", t)
}

// Stop implements the TargetProvider interface.
func (sd *ServersetDiscovery) Stop() {
	log.Debugf("Stopping serverset service discovery for %s %s", sd.conf.Servers, sd.conf.Paths)

	// Terminate Run.
	sd.runDone <- struct{}{}

	log.Debugf("Serverset service discovery for %s %s stopped", sd.conf.Servers, sd.conf.Paths)
}

func probeTCP(target string, w http.ResponseWriter, module Module) bool {
	deadline := time.Now().Add(module.Timeout)
	conn, err := net.DialTimeout("tcp", target, module.Timeout)
	if err != nil {
		return false
	}
	defer conn.Close()
	// Set a deadline to prevent the following code from blocking forever.
	// If a deadline cannot be set, better fail the probe by returning an error
	// now rather than blocking forever.
	if err := conn.SetDeadline(deadline); err != nil {
		return false
	}
	scanner := bufio.NewScanner(conn)
	for _, qr := range module.TCP.QueryResponse {
		log.Debugf("Processing query response entry %+v", qr)
		send := qr.Send
		if qr.Expect != "" {
			re, err := regexp.Compile(qr.Expect)
			if err != nil {
				log.Errorf("Could not compile %q into regular expression: %v", qr.Expect, err)
				return false
			}
			var match []int
			// Read lines until one of them matches the configured regexp.
			for scanner.Scan() {
				log.Debugf("read %q\n", scanner.Text())
				match = re.FindSubmatchIndex(scanner.Bytes())
				if match != nil {
					log.Debugf("regexp %q matched %q", re, scanner.Text())
					break
				}
			}
			if scanner.Err() != nil {
				return false
			}
			if match == nil {
				return false
			}
			send = string(re.Expand(nil, []byte(send), scanner.Bytes(), match))
		}
		if send != "" {
			log.Debugf("Sending %q", send)
			if _, err := fmt.Fprintf(conn, "%s\n", send); err != nil {
				return false
			}
		}
	}
	return true
}

func (e *exporter) scrape(ch chan<- prometheus.Metric) {
	defer close(ch)

	servers := []string{}

	if e.useExhibitor {
		url := fmt.Sprintf("http://%s/exhibitor/v1/cluster/list", e.addrs[0])
		rr, err := http.NewRequest("GET", url, nil)
		if err != nil {
			panic(err)
		}

		rresp, err := httpClient.Transport.RoundTrip(rr)
		if err != nil {
			e.recordErr(err)
			return
		}
		defer rresp.Body.Close()

		body, err := ioutil.ReadAll(rresp.Body)
		if err != nil {
			e.recordErr(err)
			return
		}

		var serverList Servers
		err = json.Unmarshal(body, &serverList)
		if err != nil {
			e.recordErr(err)
			return
		}

		log.Debugf("Got serverlist from Exhibitor: %s", serverList)

		for _, host := range serverList.Servers {
			servers = append(servers, fmt.Sprintf("%s:%d", host, serverList.Port))
		}
	} else {
		servers = e.addrs
	}

	log.Debugf("Polling servers: %s", servers)
	var wg sync.WaitGroup
	for _, server := range servers {
		log.Debugf("Polling server: %s", server)
		wg.Add(1)
		go e.pollServer(server, ch, &wg)
	}
	wg.Wait()
}

func ParseReceiver(s string) (receiver string, err error) {
	var si interface{}

	rdr := bytes.NewReader([]byte(s))
	v, err := ReadRESP(rdr)
	if err != nil {
		return _EMPTY_, err
	}
	log.Debugf("[ParseReceiver] %v: %q", reflect.TypeOf(v), v)

	switch t := v.(type) {
	case []interface{}:
		si = t[receiverPos]
	default:
		return _EMPTY_, errors.New("No receiver found.")
	}

	switch t := si.(type) {
	case interface{}:
		receiver = t.(string)
	case string:
		receiver = t
	case int:
		receiver = strconv.Itoa(t)
	}
	if receiver == _EMPTY_ {
		return receiver, errors.New("No receiver found.")
	}
	return
}

// Expose filesystem fullness.
func (c *filesystemCollector) Update(ch chan<- prometheus.Metric) (err error) {
	var mntbuf *C.struct_statfs
	count := C.getmntinfo(&mntbuf, C.MNT_NOWAIT)
	if count == 0 {
		return errors.New("getmntinfo() failed")
	}

	mnt := (*[1 << 30]C.struct_statfs)(unsafe.Pointer(mntbuf))
	for i := 0; i < int(count); i++ {
		name := C.GoString(&mnt[i].f_mntonname[0])
		if c.ignoredMountPointsPattern.MatchString(name) {
			log.Debugf("Ignoring mount point: %s", name)
			continue
		}
		c.size.WithLabelValues(name).Set(float64(mnt[i].f_blocks) * float64(mnt[i].f_bsize))
		c.free.WithLabelValues(name).Set(float64(mnt[i].f_bfree) * float64(mnt[i].f_bsize))
		c.avail.WithLabelValues(name).Set(float64(mnt[i].f_bavail) * float64(mnt[i].f_bsize))
		c.files.WithLabelValues(name).Set(float64(mnt[i].f_files))
		c.filesFree.WithLabelValues(name).Set(float64(mnt[i].f_ffree))
	}

	c.size.Collect(ch)
	c.free.Collect(ch)
	c.avail.Collect(ch)
	c.files.Collect(ch)
	c.filesFree.Collect(ch)
	return err
}

// Store sends a batch of samples to InfluxDB via its HTTP API.
func (c *Client) Store(samples model.Samples) error {
	points := make([]influx.Point, 0, len(samples))
	for _, s := range samples {
		v := float64(s.Value)
		if math.IsNaN(v) || math.IsInf(v, 0) {
			log.Debugf("cannot send value %f to InfluxDB, skipping sample %#v", v, s)
			c.ignoredSamples.Inc()
			continue
		}
		points = append(points, influx.Point{
			Measurement: string(s.Metric[model.MetricNameLabel]),
			Tags:        tagsFromMetric(s.Metric),
			Time:        s.Timestamp.Time(),
			Precision:   "ms",
			Fields: map[string]interface{}{
				"value": v,
			},
		})
	}

	bps := influx.BatchPoints{
		Points:          points,
		Database:        c.database,
		RetentionPolicy: c.retentionPolicy,
	}
	_, err := c.client.Write(bps)
	return err
}

func (tc *zookeeperTreeCache) loop(failureMode bool) {
	retryChan := make(chan struct{})

	failure := func() {
		failureMode = true
		time.AfterFunc(time.Second*10, func() {
			retryChan <- struct{}{}
		})
	}
	if failureMode {
		failure()
	}

	for {
		select {
		case ev := <-tc.head.events:
			log.Debugf("Received Zookeeper event: %s", ev)
			if failureMode {
				continue
			}
			if ev.Type == zk.EventNotWatching {
				log.Infof("Lost connection to Zookeeper.")
				failure()
			} else {
				path := strings.TrimPrefix(ev.Path, tc.prefix)
				parts := strings.Split(path, "/")
				node := tc.head
				for _, part := range parts[1:] {
					childNode := node.children[part]
					if childNode == nil {
						childNode = &zookeeperTreeCacheNode{
							events:   tc.head.events,
							children: map[string]*zookeeperTreeCacheNode{},
							done:     make(chan struct{}, 1),
						}
						node.children[part] = childNode
					}
					node = childNode
				}
				err := tc.recursiveNodeUpdate(ev.Path, node)
				if err != nil {
					log.Errorf("Error during processing of Zookeeper event: %s", err)
					failure()
				}
			}
		case <-retryChan:
			log.Infof("Attempting to resync state with Zookeeper")
			err := tc.recursiveNodeUpdate(tc.prefix, tc.head)
			if err == nil {
				failureMode = false
			} else {
				log.Errorf("Error during Zookeeper resync: %s", err)
				failure()
			}
		case <-tc.stop:
			close(tc.events)
			return
		}
	}
}

func activateHandler(w http.ResponseWriter, r *http.Request) {
	// TODO(kendall): Count activations
	// TODO(kendall): Return errors or at least clean up the logic
	// TODO(kendall): Flag for ssl?
	// TODO(kendall): Figure out host and port
	// TODO(kendall): Authtoken
	// TODO(kendall): postrate
	// TODO(kendall): highprec
	var activation ted5000ActivationRequest
	var port_suffix string
	var port int
	var err error
	if err := xml.NewDecoder(r.Body).Decode(&activation); err != nil {
		fmt.Fprintf(w, "Could not parse activation XML: %s", err)
	}
	log.Debugf("Activation request: %s", activation)
	if _, port_suffix, err = net.SplitHostPort(*listenAddress); err != nil {
		fmt.Printf("Could not determine port from %s: %s", *listenAddress, err)
	}
	if port, err = strconv.Atoi(port_suffix); err != nil {
		fmt.Printf("Could not create port (%s) to int: %s", port, err)
	}
	if err := xml.NewEncoder(w).Encode(ted5000ActivationResponse{
		PostServer: r.Host,
		UseSSL:     false,
		PostPort:   port,
		PostRate:   *postRate,
		PostURL:    "/post",
		HighPrec:   "T"}); err != nil {
		fmt.Fprintf(w, "Could not create XML activation response: %s", err)
	}
}

func ParsePayload(s string) (payload string, err error) {
	var pi interface{}

	rdr := bytes.NewReader([]byte(s))
	v, err := ReadRESP(rdr)
	if err != nil {
		return _EMPTY_, err
	}
	log.Debugf("%v: %q", reflect.TypeOf(v), v)
	switch t := v.(type) {
	case []interface{}:
		pi = t[payloadPos]
	default:
		return _EMPTY_, errors.New("No receiver found.")
	}

	switch t := pi.(type) {
	case interface{}:
		payload = t.(string)
	case string:
		payload = t
	case int:
		payload = strconv.Itoa(t)
	}
	if payload == _EMPTY_ {
		return payload, errors.New("No payload found.")
	}
	return
}

func ParseEvent(s string) (event string, err error) {
	var ei interface{}

	rdr := bytes.NewReader([]byte(s))
	v, err := ReadRESP(rdr)
	if err != nil {
		return _EMPTY_, err
	}
	log.Debugf("[ParseEvent] %v: %v", reflect.TypeOf(v), v)

	switch t := v.(type) {
	case []interface{}:
		ei = t[eventPos]
	default:
		return _EMPTY_, errors.New("No receiver found.")
	}

	switch t := ei.(type) {
	case interface{}:
		event = t.(string)
	case string:
		event = t
	case int:
		event = strconv.Itoa(t)
	}
	if event == _EMPTY_ {
		return event, errors.New("No event found.")
	}
	return
}

func (c *timeCollector) Update(ch chan<- prometheus.Metric) (err error) {
	now := float64(time.Now().Unix())
	log.Debugf("Set time: %f", now)
	c.metric.Set(now)
	c.metric.Collect(ch)
	return err
}

// Expose filesystem fullness.
func (c *filesystemCollector) Update(ch chan<- prometheus.Metric) (err error) {
	mpds, err := mountPointDetails()
	if err != nil {
		return err
	}
	for _, mpd := range mpds {
		if c.ignoredMountPointsPattern.MatchString(mpd.mountPoint) {
			log.Debugf("Ignoring mount point: %s", mpd.mountPoint)
			continue
		}
		buf := new(syscall.Statfs_t)
		err := syscall.Statfs(mpd.mountPoint, buf)
		if err != nil {
			return fmt.Errorf("Statfs on %s returned %s", mpd.mountPoint, err)
		}

		c.size.WithLabelValues(mpd.device, mpd.mountPoint, mpd.fsType).Set(float64(buf.Blocks) * float64(buf.Bsize))
		c.free.WithLabelValues(mpd.device, mpd.mountPoint, mpd.fsType).Set(float64(buf.Bfree) * float64(buf.Bsize))
		c.avail.WithLabelValues(mpd.device, mpd.mountPoint, mpd.fsType).Set(float64(buf.Bavail) * float64(buf.Bsize))
		c.files.WithLabelValues(mpd.device, mpd.mountPoint, mpd.fsType).Set(float64(buf.Files))
		c.filesFree.WithLabelValues(mpd.device, mpd.mountPoint, mpd.fsType).Set(float64(buf.Ffree))
	}
	c.size.Collect(ch)
	c.free.Collect(ch)
	c.avail.Collect(ch)
	c.files.Collect(ch)
	c.filesFree.Collect(ch)
	return err
}

// Expose filesystem fullness.
func (c *filesystemCollector) GetStats() (stats []filesystemStats, err error) {
	var mntbuf *C.struct_statfs
	count := C.getmntinfo(&mntbuf, C.MNT_NOWAIT)
	if count == 0 {
		return nil, errors.New("getmntinfo() failed")
	}

	mnt := (*[1 << 30]C.struct_statfs)(unsafe.Pointer(mntbuf))
	stats = []filesystemStats{}
	for i := 0; i < int(count); i++ {
		name := C.GoString(&mnt[i].f_mntonname[0])
		if c.ignoredMountPointsPattern.MatchString(name) {
			log.Debugf("Ignoring mount point: %s", name)
			continue
		}
		labelValues := []string{name}
		stats = append(stats, filesystemStats{
			labelValues: labelValues,
			size:        float64(mnt[i].f_blocks) * float64(mnt[i].f_bsize),
			free:        float64(mnt[i].f_bfree) * float64(mnt[i].f_bsize),
			avail:       float64(mnt[i].f_bavail) * float64(mnt[i].f_bsize),
			files:       float64(mnt[i].f_files),
			filesFree:   float64(mnt[i].f_ffree),
		})
	}
	return stats, nil
}

// Query handles the /api/query endpoint.
func (api *API) Query(w http.ResponseWriter, r *http.Request) {
	setAccessControlHeaders(w)
	w.Header().Set("Content-Type", "application/json")

	params := httputil.GetQueryParams(r)
	expr := params.Get("expr")

	timestamp, err := parseTimestampOrNow(params.Get("timestamp"), api.Now())
	if err != nil {
		httpJSONError(w, fmt.Errorf("invalid query timestamp %s", err), http.StatusBadRequest)
		return
	}

	query, err := api.QueryEngine.NewInstantQuery(expr, timestamp)
	if err != nil {
		httpJSONError(w, err, http.StatusOK)
		return
	}
	res := query.Exec()
	if res.Err != nil {
		httpJSONError(w, res.Err, http.StatusOK)
		return
	}
	log.Debugf("Instant query: %s\nQuery stats:\n%s\n", expr, query.Stats())

	httputil.RespondJSON(w, res.Value)
}

func (c *gmondCollector) Update(ch chan<- prometheus.Metric) (err error) {
	conn, err := net.Dial(gangliaProto, gangliaAddress)
	log.Debugf("gmondCollector Update")
	if err != nil {
		return fmt.Errorf("Can't connect to gmond: %s", err)
	}
	conn.SetDeadline(time.Now().Add(gangliaTimeout))

	ganglia := ganglia.Ganglia{}
	decoder := xml.NewDecoder(bufio.NewReader(conn))
	decoder.CharsetReader = toUtf8

	err = decoder.Decode(&ganglia)
	if err != nil {
		return fmt.Errorf("Couldn't parse xml: %s", err)
	}

	for _, cluster := range ganglia.Clusters {
		for _, host := range cluster.Hosts {

			for _, metric := range host.Metrics {
				name := illegalCharsRE.ReplaceAllString(metric.Name, "_")

				c.setMetric(name, cluster.Name, metric)
			}
		}
	}
	for _, m := range c.metrics {
		m.Collect(ch)
	}
	return err
}