func (c *processesCollector) Update(ch chan<- prometheus.Metric) error {
	processes, err := procfs.AllProcs()
	if err != nil {
		return fmt.Errorf("failed to get processes: %s", err)
	}

	for _, process := range processes {
		cmd, err := process.Comm()
		if err != nil {
			log.Debugf("Failed to get process command: %s", err)
			continue
		}

		stats, err := process.NewStat()
		if err != nil {
			log.Debugf("Failed to get process stats: %s", err)
			continue
		}

		// skip processes with empty stats
		if stats.ResidentMemory() == 0 || stats.CPUTime() == 0 {
			log.Debugf("Skipping process %s due to empty stats", cmd)
			continue
		}

		c.metrics[0].(*prometheus.GaugeVec).WithLabelValues(cmd).Set(float64(stats.ResidentMemory()))
		c.metrics[1].(*prometheus.GaugeVec).WithLabelValues(cmd).Set(stats.CPUTime())
	}

	for _, c := range c.metrics {
		c.Collect(ch)
	}

	return err
}

// 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
}

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)
}

func (d *serviceDiscovery) run(ctx context.Context, ch chan<- []*config.TargetGroup) {
	update := make(chan interface{}, 10)
	go d.startServiceWatch(update, ctx.Done(), d.retryInterval)

	for {
		tgs := []*config.TargetGroup{}
		select {
		case <-ctx.Done():
			return
		case event := <-update:
			switch e := event.(type) {
			case *endpointsEvent:
				log.Debugf("k8s discovery received endpoint event (EventType=%s, Endpoint Name=%s)", e.EventType, e.Endpoints.ObjectMeta.Name)
				tgs = append(tgs, d.updateServiceEndpoints(e.Endpoints, e.EventType))
			case *serviceEvent:
				log.Debugf("k8s discovery received service event (EventType=%s, Service Name=%s)", e.EventType, e.Service.ObjectMeta.Name)
				tgs = append(tgs, d.updateService(e.Service, e.EventType))
			}
		}
		if tgs == nil {
			continue
		}

		for _, tg := range tgs {
			select {
			case ch <- []*config.TargetGroup{tg}:
			case <-ctx.Done():
				return
			}
		}
	}

}

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 *runitCollector) Update(ch chan<- prometheus.Metric) error {
	services, err := runit.GetServices(*runitServiceDir)
	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)
		ch <- c.state.mustNewConstMetric(float64(status.State), service.Name)
		ch <- c.stateDesired.mustNewConstMetric(float64(status.Want), service.Name)
		ch <- c.stateTimestamp.mustNewConstMetric(float64(status.Timestamp.Unix()), service.Name)
		if status.NormallyUp {
			ch <- c.stateNormal.mustNewConstMetric(1, service.Name)
		} else {
			ch <- c.stateNormal.mustNewConstMetric(0, service.Name)
		}
	}
	return 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)
}

func probeTCP(target string, w http.ResponseWriter, module Module) bool {
	deadline := time.Now().Add(module.Timeout)
	conn, err := dialTCP(target, w, module)
	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
	}
	if module.TCP.TLS {
		state := conn.(*tls.Conn).ConnectionState()
		fmt.Fprintf(w, "probe_ssl_earliest_cert_expiry %f\n",
			float64(getEarliestCertExpiry(&state).UnixNano())/1e9)
	}
	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 (c *drbdCollector) Update(ch chan<- prometheus.Metric) (err error) {
	statsFile := procFilePath("drbd")
	file, err := os.Open(statsFile)
	if err != nil {
		if os.IsNotExist(err) {
			log.Debugf("Not collecting DRBD statistics, as %s does not exist: %s", statsFile, err)
			return nil
		}
		return err
	}
	defer file.Close()

	scanner := bufio.NewScanner(file)
	scanner.Split(bufio.ScanWords)
	device := "unknown"
	for scanner.Scan() {
		field := scanner.Text()
		if kv := strings.Split(field, ":"); len(kv) == 2 {
			if id, err := strconv.ParseUint(kv[0], 10, 64); err == nil && kv[1] == "" {
				device = fmt.Sprintf("drbd%d", id)
			} else if metric, ok := drbdNumericalMetrics[kv[0]]; ok {
				// Numerical value.
				value, err := strconv.ParseFloat(kv[1], 64)
				if err != nil {
					return err
				}
				ch <- prometheus.MustNewConstMetric(
					metric.desc, metric.valueType,
					value*metric.multiplier, device)
			} else if metric, ok := drbdStringPairMetrics[kv[0]]; ok {
				// String pair value.
				values := strings.Split(kv[1], "/")
				ch <- prometheus.MustNewConstMetric(
					metric.desc, prometheus.GaugeValue,
					metric.isOkay(values[0]), device, "local")
				ch <- prometheus.MustNewConstMetric(
					metric.desc, prometheus.GaugeValue,
					metric.isOkay(values[1]), device, "remote")
			} else if kv[0] == "cs" {
				// Connection state.
				var connected float64
				if kv[1] == "Connected" {
					connected = 1
				}
				ch <- prometheus.MustNewConstMetric(
					drbdConnected, prometheus.GaugeValue,
					connected, device)
			} else {
				log.Debugf("Don't know how to process key-value pair [%s: %q]", kv[0], kv[1])
			}
		} else {
			log.Debugf("Don't know how to process string %q", field)
		}
	}
	return scanner.Err()
}

func (c *ntpCollector) Update(ch chan<- prometheus.Metric) (err error) {
	resp, err := ntp.Query(*ntpServer, *ntpProtocolVersion)
	if err != nil {
		return fmt.Errorf("couldn't get NTP drift: %s", err)
	}
	driftSeconds := resp.ClockOffset.Seconds()
	log.Debugf("Set ntp_drift_seconds: %f", driftSeconds)
	ch <- c.drift.mustNewConstMetric(driftSeconds)

	stratum := float64(resp.Stratum)
	log.Debugf("Set ntp_stratum: %f", stratum)
	ch <- c.stratum.mustNewConstMetric(stratum)
	return nil
}

func (e *Exporter) scrapeApps(json *gabs.Container, ch chan<- prometheus.Metric) {
	elements, _ := json.S("apps").Children()
	states := map[string]string{
		"running":    "tasksRunning",
		"staged":     "tasksStaged",
		"healthy":    "tasksHealthy",
		"unhealthy":  "tasksUnhealthy",
		"cpus":       "cpus",
		"mem_in_mb":  "mem",
		"disk_in_mb": "disk",
		"gpus":       "gpus",
		"avg_uptime": "taskStats.startedAfterLastScaling.stats.lifeTime.averageSeconds",
	}

	name := "app_instances"
	gauge, new := e.Gauges.Fetch(name, "Marathon app instance count", "app")
	if new {
		log.Infof("Added gauge %q\n", name)
	}
	gauge.Reset()

	for _, app := range elements {
		id := app.Path("id").Data().(string)
		data := app.Path("instances").Data()
		count, ok := data.(float64)
		if !ok {
			log.Debugf(fmt.Sprintf("Bad conversion! Unexpected value \"%v\" for number of app instances\n", data))
			continue
		}

		gauge.WithLabelValues(id).Set(count)

		for key, value := range states {
			name := fmt.Sprintf("app_task_%s", key)
			gauge, new := e.Gauges.Fetch(name, fmt.Sprintf("Marathon app task %s count", key), "app")
			if new {
				log.Infof("Added gauge %q\n", name)
			}

			data := app.Path(value).Data()
			count, ok := data.(float64)
			if !ok {
				log.Debugf(fmt.Sprintf("Bad conversion! Unexpected value \"%v\" for number of \"%s\" tasks\n", data, key))
				continue
			}

			gauge.WithLabelValues(id).Set(count)
		}
	}
}

// Expose filesystem fullness.
func (c *filesystemCollector) GetStats() (stats []filesystemStats, err error) {
	buf := make([]unix.Statfs_t, 16)
	for {
		n, err := unix.Getfsstat(buf, MNT_NOWAIT)
		if err != nil {
			return nil, err
		}
		if n < len(buf) {
			buf = buf[:n]
			break
		}
		buf = make([]unix.Statfs_t, len(buf)*2)
	}
	stats = []filesystemStats{}
	for _, fs := range buf {
		mountpoint := gostring(fs.Mntonname[:])
		if c.ignoredMountPointsPattern.MatchString(mountpoint) {
			log.Debugf("Ignoring mount point: %s", mountpoint)
			continue
		}

		device := gostring(fs.Mntfromname[:])
		fstype := gostring(fs.Fstypename[:])
		if c.ignoredFSTypesPattern.MatchString(fstype) {
			log.Debugf("Ignoring fs type: %s", fstype)
			continue
		}

		var ro float64
		if (fs.Flags & MNT_RDONLY) != 0 {
			ro = 1
		}

		stats = append(stats, filesystemStats{
			labels: filesystemLabels{
				device:     device,
				mountPoint: mountpoint,
				fsType:     fstype,
			},
			size:      float64(fs.Blocks) * float64(fs.Bsize),
			free:      float64(fs.Bfree) * float64(fs.Bsize),
			avail:     float64(fs.Bavail) * float64(fs.Bsize),
			files:     float64(fs.Files),
			filesFree: float64(fs.Ffree),
			ro:        ro,
		})
	}
	return stats, nil
}

// 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++ {
		mountpoint := C.GoString(&mnt[i].f_mntonname[0])
		if c.ignoredMountPointsPattern.MatchString(mountpoint) {
			log.Debugf("Ignoring mount point: %s", mountpoint)
			continue
		}

		device := C.GoString(&mnt[i].f_mntfromname[0])
		fstype := C.GoString(&mnt[i].f_fstypename[0])

		labelValues := []string{device, mountpoint, fstype}
		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
}

func (d *nodeDiscovery) run(ctx context.Context, ch chan<- []*config.TargetGroup) {
	select {
	case ch <- []*config.TargetGroup{d.updateNodesTargetGroup()}:
	case <-ctx.Done():
		return
	}

	update := make(chan *nodeEvent, 10)
	go d.watchNodes(update, ctx.Done(), d.retryInterval)

	for {
		tgs := []*config.TargetGroup{}
		select {
		case <-ctx.Done():
			return
		case e := <-update:
			log.Debugf("k8s discovery received node event (EventType=%s, Node Name=%s)", e.EventType, e.Node.ObjectMeta.Name)
			d.updateNode(e.Node, e.EventType)
			tgs = append(tgs, d.updateNodesTargetGroup())
		}
		if tgs == nil {
			continue
		}

		for _, tg := range tgs {
			select {
			case ch <- []*config.TargetGroup{tg}:
			case <-ctx.Done():
				return
			}
		}
	}
}

func (md *Discovery) updateServices(ctx context.Context, ch chan<- []*config.TargetGroup) error {
	targetMap, err := md.fetchTargetGroups()
	if err != nil {
		return err
	}

	all := make([]*config.TargetGroup, 0, len(targetMap))
	for _, tg := range targetMap {
		all = append(all, tg)
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	case ch <- all:
	}

	// Remove services which did disappear.
	for source := range md.lastRefresh {
		_, ok := targetMap[source]
		if !ok {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case ch <- []*config.TargetGroup{{Source: source}}:
				log.Debugf("Removing group for %s", source)
			}
		}
	}

	md.lastRefresh = targetMap
	return nil
}

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
}

func (c *diskstatsCollector) Update(ch chan<- prometheus.Metric) error {
	procDiskStats := procFilePath("diskstats")
	diskStats, err := getDiskStats()
	if err != nil {
		return fmt.Errorf("couldn't get diskstats: %s", err)
	}

	for dev, stats := range diskStats {
		if c.ignoredDevicesPattern.MatchString(dev) {
			log.Debugf("Ignoring device: %s", dev)
			continue
		}

		if len(stats) != len(c.descs) {
			return fmt.Errorf("invalid line for %s for %s", procDiskStats, dev)
		}

		for i, value := range stats {
			v, err := strconv.ParseFloat(value, 64)
			if err != nil {
				return fmt.Errorf("invalid value %s in diskstats: %s", value, err)
			}
			ch <- c.descs[i].mustNewConstMetric(v, dev)
		}
	}
	return nil
}

func (c *zfsCollector) parseArcstatsSysctlOutput(reader io.Reader, handler func(zfsSysctl, zfsMetricValue)) (err error) {

	// Decode values
	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {

		fields := strings.Fields(scanner.Text())

		if len(fields) != 2 ||
			!strings.HasPrefix(fields[0], zfsArcstatsSysctl) ||
			!strings.HasSuffix(fields[0], ":") {

			log.Debugf("Skipping line of unknown format: %q", scanner.Text())
			continue

		}

		sysctl := zfsSysctl(strings.TrimSuffix(fields[0], ":"))
		value, err := strconv.Atoi(fields[1])
		if err != nil {
			return err
		}

		handler(sysctl, zfsMetricValue(value))
	}
	return scanner.Err()

}

func (kd *Discovery) updateNodesTargetGroup() *config.TargetGroup {
	kd.nodesMu.RLock()
	defer kd.nodesMu.RUnlock()

	tg := &config.TargetGroup{
		Source: nodesTargetGroupName,
		Labels: model.LabelSet{
			roleLabel: model.LabelValue("node"),
		},
	}

	// Now let's loop through the nodes & add them to the target group with appropriate labels.
	for nodeName, node := range kd.nodes {
		nodeAddress, err := nodeHostIP(node)
		if err != nil {
			log.Debugf("Skipping node %s: %s", node.Name, err)
			continue
		}

		address := fmt.Sprintf("%s:%d", nodeAddress.String(), kd.Conf.KubeletPort)

		t := model.LabelSet{
			model.AddressLabel:  model.LabelValue(address),
			model.InstanceLabel: model.LabelValue(nodeName),
		}
		for k, v := range node.ObjectMeta.Labels {
			labelName := strutil.SanitizeLabelName(nodeLabelPrefix + k)
			t[model.LabelName(labelName)] = model.LabelValue(v)
		}
		tg.Targets = append(tg.Targets, t)
	}

	return tg
}

// RunScraper implements Target.
func (t *Target) RunScraper(sampleAppender storage.SampleAppender) {
	defer close(t.scraperStopped)

	lastScrapeInterval := t.interval()

	log.Debugf("Starting scraper for target %v...", t)

	select {
	case <-time.After(t.offset(lastScrapeInterval)):
		// Continue after scraping offset.
	case <-t.scraperStopping:
		return
	}

	ticker := time.NewTicker(lastScrapeInterval)
	defer ticker.Stop()

	t.scrape(sampleAppender)

	// Explanation of the contraption below:
	//
	// In case t.scraperStopping has something to receive, we want to read
	// from that channel rather than starting a new scrape (which might take very
	// long). That's why the outer select has no ticker.C. Should t.scraperStopping
	// not have anything to receive, we go into the inner select, where ticker.C
	// is in the mix.
	for {
		select {
		case <-t.scraperStopping:
			return
		default:
			select {
			case <-t.scraperStopping:
				return
			case <-ticker.C:
				took := time.Since(t.status.LastScrape())

				intervalStr := lastScrapeInterval.String()

				// On changed scrape interval the new interval becomes effective
				// after the next scrape.
				if iv := t.interval(); iv != lastScrapeInterval {
					ticker.Stop()
					ticker = time.NewTicker(iv)
					lastScrapeInterval = iv
				}

				targetIntervalLength.WithLabelValues(intervalStr).Observe(
					float64(took) / float64(time.Second), // Sub-second precision.
				)
				if sampleAppender.NeedsThrottling() {
					targetSkippedScrapes.WithLabelValues(intervalStr).Inc()
					t.status.setLastError(errSkippedScrape)
					continue
				}
				t.scrape(sampleAppender)
			}
		}
	}
}