instrument k8s-prometheus-adapter to expose prometheus metrics

fixes #218

README.md

@@ -58,6 +58,10 @@ adapter talks to Prometheus and the main Kubernetes cluster:
   metrics in the custom metrics API.  More information about this file can be found in
   [docs/config.md](docs/config.md).
 
+- `--metrics-port=N`: This changes the port on which k8s-prometheus-adapter exposes
+  metrics about its own operations. The default is port `9593`. Metrics are exposed
+  on this port at `/metrics`.
+
 Presentation
 ------------
 

cmd/adapter/adapter.go

@@ -21,6 +21,7 @@ import (
 	"crypto/x509"
 	"flag"
 	"fmt"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"io/ioutil"
 	"net/http"
 	"net/url"
@@ -65,6 +66,8 @@ type PrometheusAdapter struct {
 	MetricsRelistInterval time.Duration
 	// MetricsMaxAge is the period to query available metrics for
 	MetricsMaxAge time.Duration
+	// MetricsPort is the port on which the adapter itself will expose metrics
+	MetricsPort int16
 
 	metricsConfig *adaptercfg.MetricsDiscoveryConfig
 }
@@ -124,6 +127,8 @@ func (cmd *PrometheusAdapter) addFlags() {
 		"interval at which to re-list the set of all available metrics from Prometheus")
 	cmd.Flags().DurationVar(&cmd.MetricsMaxAge, "metrics-max-age", cmd.MetricsMaxAge, ""+
 		"period for which to query the set of available metrics from Prometheus")
+	cmd.Flags().Int16Var(&cmd.MetricsPort, "metrics-port", 9593, "port on which to expose prometheus "+
+		"metrics about k8s-prometheus-adapter")
 }
 
 func (cmd *PrometheusAdapter) loadConfig() error {
@@ -234,6 +239,14 @@ func (cmd *PrometheusAdapter) addResourceMetricsAPI(promClient prom.Client) erro
 	return nil
 }
 
+func (cmd *PrometheusAdapter) runMetrics() {
+	go func() {
+		mux := http.NewServeMux()
+		mux.Handle("/metrics", promhttp.Handler())
+		klog.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", cmd.MetricsPort), mux))
+	}()
+}
+
 func main() {
 	logs.InitLogs()
 	defer logs.FlushLogs()
@@ -289,6 +302,8 @@ func main() {
 		klog.Fatalf("unable to install resource metrics API: %v", err)
 	}
 
+	cmd.runMetrics()
+
 	// run the server
 	if err := cmd.Run(wait.NeverStop); err != nil {
 		klog.Fatalf("unable to run custom metrics adapter: %v", err)

pkg/config/loader.go

@@ -2,10 +2,11 @@ package config
 
 import (
 	"fmt"
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/metrics"
 	"io/ioutil"
 	"os"
 
-	yaml "gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v2"
 )
 
 // FromFile loads the configuration from a particular file.
@@ -28,5 +29,7 @@ func FromYAML(contents []byte) (*MetricsDiscoveryConfig, error) {
 	if err := yaml.UnmarshalStrict(contents, &cfg); err != nil {
 		return nil, fmt.Errorf("unable to parse metrics discovery config: %v", err)
 	}
+	metrics.Rules.WithLabelValues("normal").Set(float64(len(cfg.Rules)))
+	metrics.Rules.WithLabelValues("external").Set(float64(len(cfg.ExternalRules)))
 	return &cfg, nil
 }

pkg/custom-provider/provider.go

@@ -19,12 +19,13 @@ package provider
 import (
 	"context"
 	"fmt"
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/errors"
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/metrics"
 	"time"
 
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider"
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider/helpers"
 	pmodel "github.com/prometheus/common/model"
-	apierr "k8s.io/apimachinery/pkg/api/errors"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -64,6 +65,7 @@ func NewPrometheusProvider(mapper apimeta.RESTMapper, kubeClient dynamic.Interfa
 		namers:         namers,
 
 		SeriesRegistry: &basicSeriesRegistry{
+			name:   "custom",
 			mapper: mapper,
 		},
 	}
@@ -97,7 +99,7 @@ func (p *prometheusProvider) metricFor(value pmodel.SampleValue, name types.Name
 func (p *prometheusProvider) metricsFor(valueSet pmodel.Vector, info provider.CustomMetricInfo, namespace string, names []string) (*custom_metrics.MetricValueList, error) {
 	values, found := p.MatchValuesToNames(info, valueSet)
 	if !found {
-		return nil, provider.NewMetricNotFoundError(info.GroupResource, info.Metric)
+		return nil, errors.NewMetricNotFoundError(info.GroupResource, info.Metric)
 	}
 	res := []custom_metrics.MetricValue{}
 
@@ -121,7 +123,7 @@ func (p *prometheusProvider) metricsFor(valueSet pmodel.Vector, info provider.Cu
 func (p *prometheusProvider) buildQuery(info provider.CustomMetricInfo, namespace string, names ...string) (pmodel.Vector, error) {
 	query, found := p.QueryForMetric(info, namespace, names...)
 	if !found {
-		return nil, provider.NewMetricNotFoundError(info.GroupResource, info.Metric)
+		return nil, errors.NewMetricNotFoundError(info.GroupResource, info.Metric)
 	}
 
 	// TODO: use an actual context
@@ -129,12 +131,12 @@ func (p *prometheusProvider) buildQuery(info provider.CustomMetricInfo, namespac
 	if err != nil {
 		klog.Errorf("unable to fetch metrics from prometheus: %v", err)
 		// don't leak implementation details to the user
-		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
+		return nil, errors.NewInternalError(fmt.Errorf("unable to fetch metrics"))
 	}
 
 	if queryResults.Type != pmodel.ValVector {
 		klog.Errorf("unexpected results from prometheus: expected %s, got %s on results %v", pmodel.ValVector, queryResults.Type, queryResults)
-		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
+		return nil, errors.NewInternalError(fmt.Errorf("unable to fetch metrics"))
 	}
 
 	return *queryResults.Vector, nil
@@ -154,7 +156,7 @@ func (p *prometheusProvider) GetMetricByName(name types.NamespacedName, info pro
 
 	namedValues, found := p.MatchValuesToNames(info, queryResults)
 	if !found {
-		return nil, provider.NewMetricNotFoundError(info.GroupResource, info.Metric)
+		return nil, errors.NewMetricNotFoundError(info.GroupResource, info.Metric)
 	}
 
 	if len(namedValues) > 1 {
@@ -177,7 +179,7 @@ func (p *prometheusProvider) GetMetricBySelector(namespace string, selector labe
 	if err != nil {
 		klog.Errorf("unable to list matching resource names: %v", err)
 		// don't leak implementation details to the user
-		return nil, apierr.NewInternalError(fmt.Errorf("unable to list matching resources"))
+		return nil, errors.NewInternalError(fmt.Errorf("unable to list matching resources"))
 	}
 
 	// construct the actual query
@@ -252,6 +254,7 @@ func (l *cachingMetricsLister) updateMetrics() error {
 	// iterate through, blocking until we've got all results
 	for range l.namers {
 		if err := <-errs; err != nil {
+			metrics.PrometheusUp.Set(0)
 			return fmt.Errorf("unable to update list of all metrics: %v", err)
 		}
 		if ss := <-selectorSeriesChan; ss.series != nil {
@@ -259,6 +262,7 @@ func (l *cachingMetricsLister) updateMetrics() error {
 		}
 	}
 	close(errs)
+	metrics.PrometheusUp.Set(1)
 
 	newSeries := make([][]prom.Series, len(l.namers))
 	for i, namer := range l.namers {

pkg/custom-provider/series_registry.go

@@ -18,6 +18,7 @@ package provider
 
 import (
 	"fmt"
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/metrics"
 	"sync"
 
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider"
@@ -66,6 +67,9 @@ type seriesInfo struct {
 
 // overridableSeriesRegistry is a basic SeriesRegistry
 type basicSeriesRegistry struct {
+	// registry name is used for metrics &c
+	name string
+
 	mu sync.RWMutex
 
 	// info maps metric info to information about the corresponding series
@@ -122,6 +126,7 @@ func (r *basicSeriesRegistry) SetSeries(newSeriesSlices [][]prom.Series, namers
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	metrics.RegistryMetrics.WithLabelValues(r.name).Set(float64(len(newMetrics)))
 	r.info = newInfo
 	r.metrics = newMetrics
 

pkg/errors/errors.go

@@ -0,0 +1,18 @@
+package errors
+
+import (
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/metrics"
+	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider"
+	apierr "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func NewMetricNotFoundError(resource schema.GroupResource, metricName string) error {
+	metrics.Errors.WithLabelValues("not_found").Inc()
+	return provider.NewMetricNotFoundError(resource, metricName)
+}
+
+func NewInternalError(err error) *apierr.StatusError {
+	metrics.Errors.WithLabelValues("internal").Inc()
+	return apierr.NewInternalError(err)
+}

pkg/external-provider/external_series_registry.go

@@ -14,6 +14,7 @@ limitations under the License.
 package provider
 
 import (
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/metrics"
 	"sync"
 
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider"
@@ -34,6 +35,9 @@ type ExternalSeriesRegistry interface {
 
 // overridableSeriesRegistry is a basic SeriesRegistry
 type externalSeriesRegistry struct {
+	// registry name is used for metrics &c
+	name string
+
 	// We lock when reading/writing metrics, and metricsInfo to prevent inconsistencies.
 	mu sync.RWMutex
 
@@ -100,6 +104,7 @@ func (r *externalSeriesRegistry) filterAndStoreMetrics(result MetricUpdateResult
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	metrics.RegistryMetrics.WithLabelValues(r.name).Set(float64(len(apiMetricsCache)))
 	r.metrics = apiMetricsCache
 	r.metricsInfo = rawMetricsCache
 

pkg/external-provider/provider.go

@@ -16,6 +16,7 @@ package provider
 import (
 	"context"
 	"fmt"
+	"github.com/directxman12/k8s-prometheus-adapter/pkg/errors"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -25,7 +26,6 @@ import (
 
 	"github.com/kubernetes-incubator/custom-metrics-apiserver/pkg/provider"
 
-	apierr "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/metrics/pkg/apis/external_metrics"
 
@@ -45,11 +45,11 @@ func (p *externalPrometheusProvider) GetExternalMetric(namespace string, metricS
 
 	if err != nil {
 		klog.Errorf("unable to generate a query for the metric: %v", err)
-		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
+		return nil, errors.NewInternalError(fmt.Errorf("unable to fetch metrics"))
 	}
 
 	if !found {
-		return nil, provider.NewMetricNotFoundError(p.selectGroupResource(namespace), info.Metric)
+		return nil, errors.NewMetricNotFoundError(p.selectGroupResource(namespace), info.Metric)
 	}
 	// Here is where we're making the query, need to be before here xD
 	queryResults, err := p.promClient.Query(context.TODO(), pmodel.Now(), selector)
@@ -57,7 +57,7 @@ func (p *externalPrometheusProvider) GetExternalMetric(namespace string, metricS
 	if err != nil {
 		klog.Errorf("unable to fetch metrics from prometheus: %v", err)
 		// don't leak implementation details to the user
-		return nil, apierr.NewInternalError(fmt.Errorf("unable to fetch metrics"))
+		return nil, errors.NewInternalError(fmt.Errorf("unable to fetch metrics"))
 	}
 	return p.metricConverter.Convert(info, queryResults)
 }

pkg/metrics/metrics.go

@@ -0,0 +1,38 @@
+package metrics
+
+import "github.com/prometheus/client_golang/prometheus"
+
+const MetricsNamespace = "adapter"
+
+var PrometheusUp = prometheus.NewGauge(prometheus.GaugeOpts{
+	Namespace: MetricsNamespace,
+	Name:      "prometheus_up",
+	Help:      "1 when adapter is able to reach prometheus, 0 otherwise",
+})
+
+var RegistryMetrics = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Namespace: MetricsNamespace,
+	Name:      "registry_metrics",
+	Help:      "number of metrics entries in cache registry",
+}, []string{"registry"})
+
+var Errors = prometheus.NewCounterVec(prometheus.CounterOpts{
+	Namespace: MetricsNamespace,
+	Name:      "errors_total",
+	Help:      "number of errors served",
+}, []string{"type"})
+
+var Rules = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Namespace: MetricsNamespace,
+	Name:      "roles",
+	Help:      "number of configured rules",
+}, []string{"type"})
+
+func init() {
+	prometheus.MustRegister(
+		PrometheusUp,
+		RegistryMetrics,
+		Errors,
+		Rules,
+	)
+}

vendor/github.com/prometheus/client_golang/prometheus/promhttp/http.go

@@ -0,0 +1,201 @@
+// Copyright 2016 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Copyright (c) 2013, The Prometheus Authors
+// All rights reserved.
+//
+// Use of this source code is governed by a BSD-style license that can be found
+// in the LICENSE file.
+
+// Package promhttp contains functions to create http.Handler instances to
+// expose Prometheus metrics via HTTP. In later versions of this package, it
+// will also contain tooling to instrument instances of http.Handler and
+// http.RoundTripper.
+//
+// promhttp.Handler acts on the prometheus.DefaultGatherer. With HandlerFor,
+// you can create a handler for a custom registry or anything that implements
+// the Gatherer interface. It also allows to create handlers that act
+// differently on errors or allow to log errors.
+package promhttp
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+
+	"github.com/prometheus/common/expfmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	contentTypeHeader     = "Content-Type"
+	contentLengthHeader   = "Content-Length"
+	contentEncodingHeader = "Content-Encoding"
+	acceptEncodingHeader  = "Accept-Encoding"
+)
+
+var bufPool sync.Pool
+
+func getBuf() *bytes.Buffer {
+	buf := bufPool.Get()
+	if buf == nil {
+		return &bytes.Buffer{}
+	}
+	return buf.(*bytes.Buffer)
+}
+
+func giveBuf(buf *bytes.Buffer) {
+	buf.Reset()
+	bufPool.Put(buf)
+}
+
+// Handler returns an HTTP handler for the prometheus.DefaultGatherer. The
+// Handler uses the default HandlerOpts, i.e. report the first error as an HTTP
+// error, no error logging, and compression if requested by the client.
+//
+// If you want to create a Handler for the DefaultGatherer with different
+// HandlerOpts, create it with HandlerFor with prometheus.DefaultGatherer and
+// your desired HandlerOpts.
+func Handler() http.Handler {
+	return HandlerFor(prometheus.DefaultGatherer, HandlerOpts{})
+}
+
+// HandlerFor returns an http.Handler for the provided Gatherer. The behavior
+// of the Handler is defined by the provided HandlerOpts.
+func HandlerFor(reg prometheus.Gatherer, opts HandlerOpts) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		mfs, err := reg.Gather()
+		if err != nil {
+			if opts.ErrorLog != nil {
+				opts.ErrorLog.Println("error gathering metrics:", err)
+			}
+			switch opts.ErrorHandling {
+			case PanicOnError:
+				panic(err)
+			case ContinueOnError:
+				if len(mfs) == 0 {
+					http.Error(w, "No metrics gathered, last error:\n\n"+err.Error(), http.StatusInternalServerError)
+					return
+				}
+			case HTTPErrorOnError:
+				http.Error(w, "An error has occurred during metrics gathering:\n\n"+err.Error(), http.StatusInternalServerError)
+				return
+			}
+		}
+
+		contentType := expfmt.Negotiate(req.Header)
+		buf := getBuf()
+		defer giveBuf(buf)
+		writer, encoding := decorateWriter(req, buf, opts.DisableCompression)
+		enc := expfmt.NewEncoder(writer, contentType)
+		var lastErr error
+		for _, mf := range mfs {
+			if err := enc.Encode(mf); err != nil {
+				lastErr = err
+				if opts.ErrorLog != nil {
+					opts.ErrorLog.Println("error encoding metric family:", err)
+				}
+				switch opts.ErrorHandling {
+				case PanicOnError:
+					panic(err)
+				case ContinueOnError:
+					// Handled later.
+				case HTTPErrorOnError:
+					http.Error(w, "An error has occurred during metrics encoding:\n\n"+err.Error(), http.StatusInternalServerError)
+					return
+				}
+			}
+		}
+		if closer, ok := writer.(io.Closer); ok {
+			closer.Close()
+		}
+		if lastErr != nil && buf.Len() == 0 {
+			http.Error(w, "No metrics encoded, last error:\n\n"+err.Error(), http.StatusInternalServerError)
+			return
+		}
+		header := w.Header()
+		header.Set(contentTypeHeader, string(contentType))
+		header.Set(contentLengthHeader, fmt.Sprint(buf.Len()))
+		if encoding != "" {
+			header.Set(contentEncodingHeader, encoding)
+		}
+		w.Write(buf.Bytes())
+		// TODO(beorn7): Consider streaming serving of metrics.
+	})
+}
+
+// HandlerErrorHandling defines how a Handler serving metrics will handle
+// errors.
+type HandlerErrorHandling int
+
+// These constants cause handlers serving metrics to behave as described if
+// errors are encountered.
+const (
+	// Serve an HTTP status code 500 upon the first error
+	// encountered. Report the error message in the body.
+	HTTPErrorOnError HandlerErrorHandling = iota
+	// Ignore errors and try to serve as many metrics as possible.  However,
+	// if no metrics can be served, serve an HTTP status code 500 and the
+	// last error message in the body. Only use this in deliberate "best
+	// effort" metrics collection scenarios. It is recommended to at least
+	// log errors (by providing an ErrorLog in HandlerOpts) to not mask
+	// errors completely.
+	ContinueOnError
+	// Panic upon the first error encountered (useful for "crash only" apps).
+	PanicOnError
+)
+
+// Logger is the minimal interface HandlerOpts needs for logging. Note that
+// log.Logger from the standard library implements this interface, and it is
+// easy to implement by custom loggers, if they don't do so already anyway.
+type Logger interface {
+	Println(v ...interface{})
+}
+
+// HandlerOpts specifies options how to serve metrics via an http.Handler. The
+// zero value of HandlerOpts is a reasonable default.
+type HandlerOpts struct {
+	// ErrorLog specifies an optional logger for errors collecting and
+	// serving metrics. If nil, errors are not logged at all.
+	ErrorLog Logger
+	// ErrorHandling defines how errors are handled. Note that errors are
+	// logged regardless of the configured ErrorHandling provided ErrorLog
+	// is not nil.
+	ErrorHandling HandlerErrorHandling
+	// If DisableCompression is true, the handler will never compress the
+	// response, even if requested by the client.
+	DisableCompression bool
+}
+
+// decorateWriter wraps a writer to handle gzip compression if requested.  It
+// returns the decorated writer and the appropriate "Content-Encoding" header
+// (which is empty if no compression is enabled).
+func decorateWriter(request *http.Request, writer io.Writer, compressionDisabled bool) (io.Writer, string) {
+	if compressionDisabled {
+		return writer, ""
+	}
+	header := request.Header.Get(acceptEncodingHeader)
+	parts := strings.Split(header, ",")
+	for _, part := range parts {
+		part := strings.TrimSpace(part)
+		if part == "gzip" || strings.HasPrefix(part, "gzip;") {
+			return gzip.NewWriter(writer), "gzip"
+		}
+	}
+	return writer, ""
+}

vendor/modules.txt

@@ -136,6 +136,7 @@ github.com/pborman/uuid
 # github.com/pmezard/go-difflib v1.0.0
 github.com/pmezard/go-difflib/difflib
 # github.com/prometheus/client_golang v0.8.0
+github.com/prometheus/client_golang/prometheus/promhttp
 github.com/prometheus/client_golang/prometheus
 # github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910
 github.com/prometheus/client_model/go
@@ -291,13 +292,13 @@ k8s.io/api/authorization/v1beta1
 k8s.io/api/admission/v1beta1
 # k8s.io/apimachinery v0.0.0-20190404173353-6a84e37a896d
 k8s.io/apimachinery/pkg/util/wait
-k8s.io/apimachinery/pkg/api/errors
 k8s.io/apimachinery/pkg/api/meta
 k8s.io/apimachinery/pkg/api/resource
 k8s.io/apimachinery/pkg/apis/meta/v1
 k8s.io/apimachinery/pkg/labels
 k8s.io/apimachinery/pkg/types
 k8s.io/apimachinery/pkg/util/runtime
+k8s.io/apimachinery/pkg/api/errors
 k8s.io/apimachinery/pkg/runtime/schema
 k8s.io/apimachinery/pkg/selection
 k8s.io/apimachinery/pkg/runtime
@@ -309,13 +310,13 @@ k8s.io/apimachinery/pkg/watch
 k8s.io/apimachinery/pkg/util/errors
 k8s.io/apimachinery/pkg/util/validation
 k8s.io/apimachinery/pkg/apis/meta/v1/unstructured
-k8s.io/apimachinery/pkg/util/validation/field
 k8s.io/apimachinery/pkg/apis/meta/v1beta1
 k8s.io/apimachinery/pkg/conversion
 k8s.io/apimachinery/pkg/fields
 k8s.io/apimachinery/pkg/util/intstr
 k8s.io/apimachinery/pkg/runtime/serializer/json
 k8s.io/apimachinery/pkg/runtime/serializer/versioning
+k8s.io/apimachinery/pkg/util/validation/field
 k8s.io/apimachinery/pkg/version
 k8s.io/apimachinery/pkg/apis/meta/internalversion
 k8s.io/apimachinery/pkg/conversion/queryparams