Simple ZMON collector implementation (#2)

* Simple ZMON collector implementation Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de> * Add tests for ZMON client Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de> * Add tests for zmon collector Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de> * Update ZMON collector docs Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de> * Expose tags instead of entities for queries Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de> * Remove unused function Signed-off-by: Mikkel Oscar Lyderik Larsen <mikkel.larsen@zalando.de>
2025-08-15 02:59:46 +00:00 · 2018-10-29 14:26:25 +01:00
parent b18acf3ed0
commit c86a82ca88
8 changed files with 964 additions and 31 deletions
--- a/pkg/collector/zmon_collector.go
+++ b/pkg/collector/zmon_collector.go
@@ -0,0 +1,175 @@
+package collector
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/zalando-incubator/kube-metrics-adapter/pkg/zmon"
+	autoscalingv2beta1 "k8s.io/api/autoscaling/v2beta1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+const (
+	// ZMONCheckMetric defines the metric name for metrics based on ZMON
+	// checks.
+	ZMONCheckMetric            = "zmon-check"
+	zmonCheckIDLabelKey        = "check-id"
+	zmonKeyLabelKey            = "key"
+	zmonDurationLabelKey       = "duration"
+	zmonAggregatorsLabelKey    = "aggregators"
+	zmonTagPrefixLabelKey      = "tag-"
+	defaultQueryDuration       = 10 * time.Minute
+	zmonKeyAnnotationKey       = "metric-config.external.zmon-check.zmon/key"
+	zmonTagPrefixAnnotationKey = "metric-config.external.zmon-check.zmon/tag-"
+)
+
+// ZMONCollectorPlugin defines a plugin for creating collectors that can get
+// metrics from ZMON.
+type ZMONCollectorPlugin struct {
+	zmon zmon.ZMON
+}
+
+// NewZMONCollectorPlugin initializes a new ZMONCollectorPlugin.
+func NewZMONCollectorPlugin(zmon zmon.ZMON) (*ZMONCollectorPlugin, error) {
+	return &ZMONCollectorPlugin{
+		zmon: zmon,
+	}, nil
+}
+
+// NewCollector initializes a new ZMON collector from the specified HPA.
+func (c *ZMONCollectorPlugin) NewCollector(hpa *autoscalingv2beta1.HorizontalPodAutoscaler, config *MetricConfig, interval time.Duration) (Collector, error) {
+	switch config.Name {
+	case ZMONCheckMetric:
+		annotations := map[string]string{}
+		if hpa != nil {
+			annotations = hpa.Annotations
+		}
+		return NewZMONCollector(c.zmon, config, annotations, interval)
+	}
+
+	return nil, fmt.Errorf("metric '%s' not supported", config.Name)
+}
+
+// ZMONCollector defines a collector that is able to collect metrics from ZMON.
+type ZMONCollector struct {
+	zmon        zmon.ZMON
+	interval    time.Duration
+	checkID     int
+	key         string
+	labels      map[string]string
+	tags        map[string]string
+	duration    time.Duration
+	aggregators []string
+	metricName  string
+	metricType  autoscalingv2beta1.MetricSourceType
+}
+
+// NewZMONCollector initializes a new ZMONCollector.
+func NewZMONCollector(zmon zmon.ZMON, config *MetricConfig, annotations map[string]string, interval time.Duration) (*ZMONCollector, error) {
+	checkIDStr, ok := config.Labels[zmonCheckIDLabelKey]
+	if !ok {
+		return nil, fmt.Errorf("ZMON check ID not specified on metric")
+	}
+
+	checkID, err := strconv.Atoi(checkIDStr)
+	if err != nil {
+		return nil, err
+	}
+
+	key := ""
+
+	// get optional key
+	if k, ok := config.Labels[zmonKeyLabelKey]; ok {
+		key = k
+	}
+
+	// annotations takes precedence over label
+	if k, ok := annotations[zmonKeyAnnotationKey]; ok {
+		key = k
+	}
+
+	duration := defaultQueryDuration
+
+	// parse optional duration value
+	if d, ok := config.Labels[zmonDurationLabelKey]; ok {
+		duration, err = time.ParseDuration(d)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// parse tags
+	tags := make(map[string]string)
+	for k, v := range config.Labels {
+		if strings.HasPrefix(k, zmonTagPrefixLabelKey) {
+			key := strings.TrimPrefix(k, zmonTagPrefixLabelKey)
+			tags[key] = v
+		}
+	}
+
+	// parse tags from annotations
+	// tags defined in annotations takes precedence over tags defined in
+	// the labels.
+	for k, v := range annotations {
+		if strings.HasPrefix(k, zmonTagPrefixAnnotationKey) {
+			key := strings.TrimPrefix(k, zmonTagPrefixAnnotationKey)
+			tags[key] = v
+		}
+	}
+
+	// default aggregator is last
+	aggregators := []string{"last"}
+	if k, ok := config.Labels[zmonAggregatorsLabelKey]; ok {
+		aggregators = strings.Split(k, ",")
+	}
+
+	return &ZMONCollector{
+		zmon:        zmon,
+		interval:    interval,
+		checkID:     checkID,
+		key:         key,
+		tags:        tags,
+		duration:    duration,
+		aggregators: aggregators,
+		metricName:  config.Name,
+		metricType:  config.Type,
+		labels:      config.Labels,
+	}, nil
+}
+
+// GetMetrics returns a list of collected metrics for the ZMON check.
+func (c *ZMONCollector) GetMetrics() ([]CollectedMetric, error) {
+	dataPoints, err := c.zmon.Query(c.checkID, c.key, c.tags, c.aggregators, c.duration)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(dataPoints) < 1 {
+		return nil, nil
+	}
+
+	// pick the last data point
+	// TODO: do more fancy aggregations here (or in the query function)
+	point := dataPoints[len(dataPoints)-1]
+
+	metricValue := CollectedMetric{
+		Type: c.metricType,
+		External: external_metrics.ExternalMetricValue{
+			MetricName:   c.metricName,
+			MetricLabels: c.labels,
+			Timestamp:    metav1.Time{Time: point.Time},
+			Value:        *resource.NewMilliQuantity(int64(point.Value*1000), resource.DecimalSI),
+		},
+	}
+
+	return []CollectedMetric{metricValue}, nil
+}
+
+// Interval returns the interval at which the collector should run.
+func (c *ZMONCollector) Interval() time.Duration {
+	return c.interval
+}
--- a/pkg/collector/zmon_collector_test.go
+++ b/pkg/collector/zmon_collector_test.go
@@ -0,0 +1,140 @@
+package collector
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/zalando-incubator/kube-metrics-adapter/pkg/zmon"
+	autoscalingv2beta1 "k8s.io/api/autoscaling/v2beta1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+)
+
+type zmonMock struct {
+	dataPoints []zmon.DataPoint
+	entities   []zmon.Entity
+}
+
+func (m zmonMock) Query(checkID int, key string, tags map[string]string, aggregators []string, duration time.Duration) ([]zmon.DataPoint, error) {
+	return m.dataPoints, nil
+}
+
+func TestZMONCollectorNewCollector(t *testing.T) {
+	collectPlugin, _ := NewZMONCollectorPlugin(zmonMock{})
+
+	config := &MetricConfig{
+		MetricTypeName: MetricTypeName{
+			Name: ZMONCheckMetric,
+		},
+		Labels: map[string]string{
+			zmonCheckIDLabelKey:             "1234",
+			zmonAggregatorsLabelKey:         "max",
+			zmonTagPrefixLabelKey + "alias": "cluster_alias",
+			zmonDurationLabelKey:            "5m",
+			zmonKeyLabelKey:                 "key",
+		},
+	}
+
+	hpa := &autoscalingv2beta1.HorizontalPodAutoscaler{}
+
+	collector, err := collectPlugin.NewCollector(hpa, config, 1*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, collector)
+	zmonCollector := collector.(*ZMONCollector)
+	require.Equal(t, "key", zmonCollector.key)
+	require.Equal(t, 1234, zmonCollector.checkID)
+	require.Equal(t, 1*time.Second, zmonCollector.interval)
+	require.Equal(t, 5*time.Minute, zmonCollector.duration)
+	require.Equal(t, []string{"max"}, zmonCollector.aggregators)
+	require.Equal(t, map[string]string{"alias": "cluster_alias"}, zmonCollector.tags)
+
+	// check that annotations overwrites labels
+	hpa.ObjectMeta = metav1.ObjectMeta{
+		Annotations: map[string]string{
+			zmonKeyAnnotationKey:                 "annotation_key",
+			zmonTagPrefixAnnotationKey + "alias": "cluster_alias_annotation",
+		},
+	}
+	collector, err = collectPlugin.NewCollector(hpa, config, 1*time.Second)
+	require.NoError(t, err)
+	require.NotNil(t, collector)
+	zmonCollector = collector.(*ZMONCollector)
+	require.Equal(t, "annotation_key", zmonCollector.key)
+	require.Equal(t, map[string]string{"alias": "cluster_alias_annotation"}, zmonCollector.tags)
+
+	// should fail if the metric name isn't ZMON
+	config.Name = "non-zmon-check"
+	_, err = collectPlugin.NewCollector(nil, config, 1*time.Second)
+	require.Error(t, err)
+
+	// should fail if the check id is not specified.
+	delete(config.Labels, zmonCheckIDLabelKey)
+	config.Name = ZMONCheckMetric
+	_, err = collectPlugin.NewCollector(nil, config, 1*time.Second)
+	require.Error(t, err)
+}
+
+func TestZMONCollectorGetMetrics(tt *testing.T) {
+	config := &MetricConfig{
+		MetricTypeName: MetricTypeName{
+			Name: ZMONCheckMetric,
+			Type: "foo",
+		},
+		Labels: map[string]string{
+			zmonCheckIDLabelKey:             "1234",
+			zmonAggregatorsLabelKey:         "max",
+			zmonTagPrefixLabelKey + "alias": "cluster_alias",
+			zmonDurationLabelKey:            "5m",
+			zmonKeyLabelKey:                 "key",
+		},
+	}
+
+	for _, ti := range []struct {
+		msg              string
+		dataPoints       []zmon.DataPoint
+		collectedMetrics []CollectedMetric
+	}{
+		{
+			msg: "test successfully getting metrics",
+			dataPoints: []zmon.DataPoint{
+				{
+					Time:  time.Time{},
+					Value: 1.0,
+				},
+			},
+			collectedMetrics: []CollectedMetric{
+				{
+					Type: config.Type,
+					External: external_metrics.ExternalMetricValue{
+						MetricName:   config.Name,
+						MetricLabels: config.Labels,
+						Timestamp:    metav1.Time{Time: time.Time{}},
+						Value:        *resource.NewMilliQuantity(int64(1.0)*1000, resource.DecimalSI),
+					},
+				},
+			},
+		},
+		{
+			msg: "test not getting any metrics",
+		},
+	} {
+		tt.Run(ti.msg, func(t *testing.T) {
+			z := zmonMock{
+				dataPoints: ti.dataPoints,
+			}
+
+			zmonCollector, err := NewZMONCollector(z, config, nil, 1*time.Second)
+			require.NoError(t, err)
+
+			metrics, _ := zmonCollector.GetMetrics()
+			require.Equal(t, ti.collectedMetrics, metrics)
+		})
+	}
+}
+
+func TestZMONCollectorInterval(t *testing.T) {
+	collector := ZMONCollector{interval: 1 * time.Second}
+	require.Equal(t, 1*time.Second, collector.Interval())
+}
--- a/pkg/server/start.go
+++ b/pkg/server/start.go
@@ -19,6 +19,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/http"
 	"time"

@@ -28,8 +29,11 @@ import (
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/cmd/server"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
+	"github.com/zalando-incubator/cluster-lifecycle-manager/pkg/credentials-loader/platformiam"
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/collector"
 	"github.com/zalando-incubator/kube-metrics-adapter/pkg/provider"
+	"github.com/zalando-incubator/kube-metrics-adapter/pkg/zmon"
+	"golang.org/x/oauth2"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -48,6 +52,8 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		EnableCustomMetricsAPI:            true,
 		EnableExternalMetricsAPI:          true,
 		MetricsAddress:                    ":7979",
+		ZMONTokenName:                     "zmon",
+		CredentialsDir:                    "/meta/credentials",
 	}

 	cmd := &cobra.Command{
@@ -82,6 +88,14 @@ func NewCommandStartAdapterServer(stopCh <-chan struct{}) *cobra.Command {
 		"whether to enable External Metrics API")
 	flags.StringVar(&o.PrometheusServer, "prometheus-server", o.PrometheusServer, ""+
 		"url of prometheus server to query")
+	flags.StringVar(&o.ZMONKariosDBEndpoint, "zmon-kariosdb-endpoint", o.ZMONKariosDBEndpoint, ""+
+		"url of ZMON KariosDB endpoint to query for ZMON checks")
+	flags.StringVar(&o.ZMONTokenName, "zmon-token-name", o.ZMONTokenName, ""+
+		"name of the token used to query ZMON")
+	flags.StringVar(&o.Token, "token", o.Token, ""+
+		"static oauth2 token to use when calling external services like ZMON")
+	flags.StringVar(&o.CredentialsDir, "credentials-dir", o.CredentialsDir, ""+
+		"path to the credentials dir where tokens are stored")
 	flags.BoolVar(&o.SkipperIngressMetrics, "skipper-ingress-metrics", o.SkipperIngressMetrics, ""+
 		"whether to enable skipper ingress metrics")
 	flags.BoolVar(&o.AWSExternalMetrics, "aws-external-metrics", o.AWSExternalMetrics, ""+
@@ -116,6 +130,13 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 		return fmt.Errorf("unable to construct lister client config to initialize provider: %v", err)
 	}

+	// convert stop channel to a context
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-stopCh
+		cancel()
+	}()
+
 	clientConfig.Timeout = defaultClientGOTimeout

 	client, err := kubernetes.NewForConfig(clientConfig)
@@ -153,7 +174,28 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 	// register generic pod collector
 	err = collectorFactory.RegisterPodsCollector("", collector.NewPodCollectorPlugin(client))
 	if err != nil {
-		return fmt.Errorf("failed to register skipper collector plugin: %v", err)
+		return fmt.Errorf("failed to register pod collector plugin: %v", err)
+	}
+
+	// enable ZMON based metrics
+	if o.ZMONKariosDBEndpoint != "" {
+		var tokenSource oauth2.TokenSource
+		if o.Token != "" {
+			tokenSource = oauth2.StaticTokenSource(&oauth2.Token{AccessToken: o.Token})
+		} else {
+			tokenSource = platformiam.NewTokenSource(o.ZMONTokenName, o.CredentialsDir)
+		}
+
+		httpClient := newOauth2HTTPClient(ctx, tokenSource)
+
+		zmonClient := zmon.NewZMONClient(o.ZMONKariosDBEndpoint, httpClient)
+
+		zmonPlugin, err := collector.NewZMONCollectorPlugin(zmonClient)
+		if err != nil {
+			return fmt.Errorf("failed to initialize ZMON collector plugin: %v", err)
+		}
+
+		collectorFactory.RegisterExternalCollector([]string{collector.ZMONCheckMetric}, zmonPlugin)
 	}

 	awsSessions := make(map[string]*session.Session, len(o.AWSRegions))
@@ -170,13 +212,6 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct

 	hpaProvider := provider.NewHPAProvider(client, 30*time.Second, 1*time.Minute, collectorFactory)

-	// convert stop channel to a context
-	ctx, cancel := context.WithCancel(context.Background())
-	go func() {
-		<-stopCh
-		cancel()
-	}()
-
 	go hpaProvider.Run(ctx)

 	customMetricsProvider := hpaProvider
@@ -200,6 +235,45 @@ func (o AdapterServerOptions) RunCustomMetricsAdapterServer(stopCh <-chan struct
 	return server.GenericAPIServer.PrepareRun().Run(ctx.Done())
 }

+// newInstrumentedOauth2HTTPClient creates an HTTP client with automatic oauth2
+// token injection. Additionally it will spawn a go-routine for closing idle
+// connections every 20 seconds on the http.Transport. This solves the problem
+// of re-resolving DNS when the endpoint backend changes.
+// https://github.com/golang/go/issues/23427
+func newOauth2HTTPClient(ctx context.Context, tokenSource oauth2.TokenSource) *http.Client {
+	transport := &http.Transport{
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ResponseHeaderTimeout: 10 * time.Second,
+		IdleConnTimeout:       20 * time.Second,
+		MaxIdleConns:          10,
+		MaxIdleConnsPerHost:   2,
+	}
+	go func(transport *http.Transport, duration time.Duration) {
+		for {
+			select {
+			case <-time.After(duration):
+				transport.CloseIdleConnections()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}(transport, 20*time.Second)
+
+	client := &http.Client{
+		Transport: transport,
+	}
+
+	// add HTTP client to context (this is how the oauth2 lib gets it).
+	ctx = context.WithValue(ctx, oauth2.HTTPClient, client)
+
+	// instantiate an http.Client containg the token source.
+	return oauth2.NewClient(ctx, tokenSource)
+}
+
 type AdapterServerOptions struct {
 	*server.CustomMetricsAdapterServerOptions

@@ -210,8 +284,18 @@ type AdapterServerOptions struct {
 	// EnableExternalMetricsAPI switches on sample apiserver for External Metrics API
 	EnableExternalMetricsAPI bool
 	// PrometheusServer enables prometheus queries to the specified
-	// server.
+	// server
 	PrometheusServer string
+	// ZMONKariosDBEndpoint enables ZMON check queries to the specified
+	// kariosDB endpoint
+	ZMONKariosDBEndpoint string
+	// ZMONTokenName is the name of the token used to query ZMON
+	ZMONTokenName string
+	// Token is an oauth2 token used to authenticate with services like
+	// ZMON.
+	Token string
+	// CredentialsDir is the path to the dir where tokens are stored
+	CredentialsDir string
 	// SkipperIngressMetrics switches on support for skipper ingress based
 	// metric collection.
 	SkipperIngressMetrics bool
--- a/pkg/zmon/zmon.go
+++ b/pkg/zmon/zmon.go
@@ -0,0 +1,269 @@
+package zmon
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+var (
+	// set of valid aggregators that can be used in queries
+	// https://kairosdb.github.io/docs/build/html/restapi/Aggregators.html
+	validAggregators = map[string]struct{}{
+		"avg":   struct{}{},
+		"dev":   struct{}{},
+		"count": struct{}{},
+		"first": struct{}{},
+		"last":  struct{}{},
+		"max":   struct{}{},
+		"min":   struct{}{},
+		"sum":   struct{}{},
+		"diff":  struct{}{},
+	}
+)
+
+// Entity defines a ZMON entity.
+type Entity struct {
+	ID string `json:"id"`
+}
+
+// ZMON defines an interface for talking to the ZMON API.
+type ZMON interface {
+	Query(checkID int, key string, tags map[string]string, aggregators []string, duration time.Duration) ([]DataPoint, error)
+}
+
+// Client defines client for interfacing with the ZMON API.
+type Client struct {
+	dataServiceEndpoint string
+	http                *http.Client
+}
+
+// NewZMONClient initializes a new ZMON Client.
+func NewZMONClient(dataServiceEndpoint string, client *http.Client) *Client {
+	return &Client{
+		dataServiceEndpoint: dataServiceEndpoint,
+		http:                client,
+	}
+}
+
+// DataPoint defines a single datapoint returned from a query.
+type DataPoint struct {
+	Time  time.Time
+	Value float64
+}
+
+type metricQuery struct {
+	StartRelative sampling `json:"start_relative"`
+	Metrics       []metric `json:"metrics"`
+}
+
+type sampling struct {
+	Value int64  `json:"value"`
+	Unit  string `json:"unit"`
+}
+
+type metric struct {
+	Name        string              `json:"name"`
+	Limit       int                 `json:"limit"`
+	Tags        map[string][]string `json:"tags"`
+	GroupBy     []tagGroup          `json:"group_by"`
+	Aggregators []aggregator        `json:"aggregator"`
+}
+
+type tagGroup struct {
+	Name string   `json:"name"`
+	Tags []string `json:"tags"`
+}
+
+type aggregator struct {
+	Name     string   `json:"name"`
+	Sampling sampling `json:"sampling"`
+}
+
+type queryResp struct {
+	Queries []struct {
+		Results []struct {
+			Values [][]float64 `json:"values"`
+		} `json:"results"`
+	} `json:"queries"`
+}
+
+// Query queries the ZMON KairosDB endpoint and returns the resulting list of
+// data points for the query.
+//
+// https://kairosdb.github.io/docs/build/html/restapi/QueryMetrics.html
+func (c *Client) Query(checkID int, key string, tags map[string]string, aggregators []string, duration time.Duration) ([]DataPoint, error) {
+	endpoint, err := url.Parse(c.dataServiceEndpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	// convert tags map
+	tagsSlice := make(map[string][]string, len(tags))
+	for k, v := range tags {
+		tagsSlice[k] = []string{v}
+	}
+
+	query := metricQuery{
+		StartRelative: durationToSampling(duration),
+		Metrics: []metric{
+			{
+				Name:  fmt.Sprintf("zmon.check.%d", checkID),
+				Limit: 10000, // maximum limit of ZMON
+				Tags:  tagsSlice,
+				GroupBy: []tagGroup{
+					{
+						Name: "tag",
+						Tags: []string{
+							"key",
+						},
+					},
+				},
+				Aggregators: make([]aggregator, 0, len(aggregators)),
+			},
+		},
+	}
+
+	// add aggregators
+	for _, aggregatorName := range aggregators {
+		if _, ok := validAggregators[aggregatorName]; !ok {
+			return nil, fmt.Errorf("invalid aggregator '%s'", aggregatorName)
+		}
+		query.Metrics[0].Aggregators = append(query.Metrics[0].Aggregators, aggregator{
+			Name:     aggregatorName,
+			Sampling: durationToSampling(duration),
+		})
+	}
+
+	// add key to query if defined
+	if key != "" {
+		query.Metrics[0].Tags["key"] = []string{key}
+	}
+
+	body, err := json.Marshal(&query)
+	if err != nil {
+		return nil, err
+	}
+
+	endpoint.Path += "/api/v1/datapoints/query"
+
+	req, err := http.NewRequest(http.MethodPost, endpoint.String(), bytes.NewBuffer(body))
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	d, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("[kariosdb query] unexpected response code: %d", resp.StatusCode)
+	}
+
+	var result queryResp
+	err = json.Unmarshal(d, &result)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(result.Queries) < 1 {
+		return nil, nil
+	}
+
+	if len(result.Queries[0].Results) < 1 {
+		return nil, nil
+	}
+
+	dataPoints := make([]DataPoint, 0, len(result.Queries[0].Results[0].Values))
+	for _, value := range result.Queries[0].Results[0].Values {
+		if len(value) != 2 {
+			return nil, fmt.Errorf("[kariosdb query] unexpected response data")
+		}
+		point := DataPoint{
+			Time:  time.Unix(0, int64(value[0])*1000000),
+			Value: value[1],
+		}
+		dataPoints = append(dataPoints, point)
+	}
+
+	return dataPoints, nil
+}
+
+const (
+	day   = 24 * time.Hour
+	week  = day * 7
+	month = day * 30
+	year  = day * 365
+)
+
+// durationToSampling converts a time.Duration to the sampling format expected
+// by karios db. E.g. the duration `1 * time.Hour` would be converted to:
+// sampling{
+//   Unit: "minutes",
+//   Value: 1,
+// }
+func durationToSampling(d time.Duration) sampling {
+	for _, u := range []struct {
+		Unit        string
+		Nanoseconds time.Duration
+	}{
+		{
+			Unit:        "years",
+			Nanoseconds: year,
+		},
+		{
+			Unit:        "months",
+			Nanoseconds: month,
+		},
+		{
+			Unit:        "weeks",
+			Nanoseconds: week,
+		},
+		{
+			Unit:        "days",
+			Nanoseconds: day,
+		},
+		{
+			Unit:        "hours",
+			Nanoseconds: 1 * time.Hour,
+		},
+		{
+			Unit:        "minutes",
+			Nanoseconds: 1 * time.Minute,
+		},
+		{
+			Unit:        "seconds",
+			Nanoseconds: 1 * time.Second,
+		},
+		{
+			Unit:        "milliseconds",
+			Nanoseconds: 1 * time.Millisecond,
+		},
+	} {
+		if d.Nanoseconds()/int64(u.Nanoseconds) >= 1 {
+			return sampling{
+				Unit:  u.Unit,
+				Value: int64(d.Round(u.Nanoseconds) / u.Nanoseconds),
+			}
+		}
+	}
+
+	return sampling{
+		Unit:  "milliseconds",
+		Value: 0,
+	}
+}
--- a/pkg/zmon/zmon_test.go
+++ b/pkg/zmon/zmon_test.go
@@ -0,0 +1,184 @@
+package zmon
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQuery(tt *testing.T) {
+	client := &http.Client{}
+	for _, ti := range []struct {
+		msg         string
+		duration    time.Duration
+		aggregators []string
+		status      int
+		body        string
+		err         error
+		dataPoints  []DataPoint
+		key         string
+	}{
+		{
+			msg:      "test getting back a single data point",
+			duration: 1 * time.Hour,
+			status:   http.StatusOK,
+			body: `{
+			         "queries": [
+				   {
+				     "results": [
+				       {
+					 "values": [
+					  [1539710395000,765952]
+					]
+				       }
+				     ]
+				   }
+				 ]
+			}`,
+			dataPoints: []DataPoint{
+				{
+					Time:  time.Unix(1539710395, 0),
+					Value: 765952,
+				},
+			},
+		},
+		{
+			msg:      "test getting back a single datapoint with key",
+			duration: 1 * time.Hour,
+			status:   http.StatusOK,
+			key:      "my-key",
+			body: `{
+			         "queries": [
+				   {
+				     "results": [
+				       {
+					 "values": [
+					  [1539710395000,765952]
+					]
+				       }
+				     ]
+				   }
+				 ]
+			}`,
+			dataPoints: []DataPoint{
+				{
+					Time:  time.Unix(1539710395, 0),
+					Value: 765952,
+				},
+			},
+		},
+		{
+			msg:         "test getting back a single datapoint with aggregators",
+			duration:    1 * time.Hour,
+			status:      http.StatusOK,
+			aggregators: []string{"max"},
+			body: `{
+			         "queries": [
+				   {
+				     "results": [
+				       {
+					 "values": [
+					  [1539710395000,765952]
+					]
+				       }
+				     ]
+				   }
+				 ]
+			}`,
+			dataPoints: []DataPoint{
+				{
+					Time:  time.Unix(1539710395, 0),
+					Value: 765952,
+				},
+			},
+		},
+		{
+			msg:         "test query with invalid aggregator",
+			aggregators: []string{"invalid"},
+			err:         fmt.Errorf("invalid aggregator 'invalid'"),
+		},
+		{
+			msg:    "test query with invalid response",
+			status: http.StatusInternalServerError,
+			body:   `{"error": 500}`,
+			err:    fmt.Errorf("[kariosdb query] unexpected response code: 500"),
+		},
+		{
+			msg:      "test getting invalid values response",
+			duration: 1 * time.Hour,
+			status:   http.StatusOK,
+			body: `{
+			         "queries": [
+				   {
+				     "results": [
+				       {
+					 "values": [
+					  [1539710395000,765952,1]
+					]
+				       }
+				     ]
+				   }
+				 ]
+			}`,
+			err: fmt.Errorf("[kariosdb query] unexpected response data"),
+		},
+	} {
+		tt.Run(ti.msg, func(t *testing.T) {
+			ts := httptest.NewServer(http.HandlerFunc(
+				func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(ti.status)
+					w.Write([]byte(ti.body))
+				}),
+			)
+			defer ts.Close()
+
+			zmonClient := NewZMONClient(ts.URL, client)
+			dataPoints, err := zmonClient.Query(1, ti.key, nil, ti.aggregators, ti.duration)
+			assert.Equal(t, ti.err, err)
+			assert.Len(t, dataPoints, len(ti.dataPoints))
+			assert.Equal(t, ti.dataPoints, dataPoints)
+		})
+	}
+
+}
+
+func TestDurationToSampling(tt *testing.T) {
+	for _, ti := range []struct {
+		msg      string
+		duration time.Duration
+		sampling sampling
+	}{
+		{
+			msg:      "1 hour should map to hours sampling",
+			duration: 1 * time.Hour,
+			sampling: sampling{
+				Unit:  "hours",
+				Value: 1,
+			},
+		},
+		{
+			msg:      "2 years should map to years sampling",
+			duration: 2 * day * 365,
+			sampling: sampling{
+				Unit:  "years",
+				Value: 2,
+			},
+		},
+		{
+			msg:      "1 nanosecond should map to 0 milliseconds sampling",
+			duration: 1,
+			sampling: sampling{
+				Unit:  "milliseconds",
+				Value: 0,
+			},
+		},
+	} {
+		tt.Run(ti.msg, func(t *testing.T) {
+			assert.Equal(t, durationToSampling(ti.duration), ti.sampling)
+		})
+	}
+}