Merge pull request #219 from loburm/kubelet-to-gcm-metrics

Add basic metrics to kubelet-to-gcm.

kubelet-to-gcm/Makefile

@@ -15,10 +15,13 @@
 OUT_DIR = build
 PACKAGE = github.com/GoogleCloudPlatform/k8s-stackdriver/kubelet-to-gcm
 PREFIX = staging-k8s.gcr.io
-TAG = 1.2.7
+TAG = 1.2.8
 
 # Rules for building the real image for deployment to gcr.io
 
+deps:
+	dep ensure
+
 compile: monitor/kubelet monitor/controller deps
 		GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -a -o $(OUT_DIR)/monitor monitor/main/daemon.go
 

kubelet-to-gcm/monitor/main/daemon.go

@@ -18,6 +18,8 @@ package main
 
 import (
 	"flag"
+	"fmt"
+	"net/http"
 	"os"
 	"time"
 
@@ -31,6 +33,7 @@ import (
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/kubelet-to-gcm/monitor/config"
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/kubelet-to-gcm/monitor/controller"
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/kubelet-to-gcm/monitor/kubelet"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 const (
@@ -50,6 +53,7 @@ var (
 	// Flags to control runtime behavior.
 	res         = flag.Uint("resolution", 10, "The time, in seconds, to poll the Kubelet.")
 	gcmEndpoint = flag.String("gcm-endpoint", "", "The GCM endpoint to hit. Defaults to the default endpoint.")
+	port        = flag.Uint("port", 6062, "Port number used to expose metrics.")
 )
 
 func main() {
@@ -96,6 +100,11 @@ func main() {
 	}
 	log.Infof("Using GCM endpoint %q", service.BasePath)
 
+	go func() {
+		http.Handle("/metrics", promhttp.Handler())
+		log.Error(http.ListenAndServe(fmt.Sprintf(":%d", *port), nil))
+	}()
+
 	for {
 		go monitor.Once(kubeletSrc, service)
 		go monitor.Once(ctrlSrc, service)

kubelet-to-gcm/monitor/metrics.go

@@ -0,0 +1,74 @@
+/*
+Copyright 2018 Google Inc.
+
+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.
+*/
+
+package monitor
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	successfullScrapes = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "successfull_scrapes_total",
+			Help: "Number of successfull scrapes of metrics from the endpoint",
+		},
+		[]string{"source"},
+	)
+	failedScrapes = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "failed_scrapes_total",
+			Help: "Number of failed scrapes of metrics from the endpoint",
+		},
+		[]string{"source"},
+	)
+	timeseriesPushed = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "timeseries_pushed_total",
+			Help: "Number of timeseries successfully pushed to the Stackdriver",
+		},
+	)
+
+	timeseriesDropped = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "timeseries_dropped_total",
+			Help: "Number of timeseries dropped during a push to the Stackdriver",
+		},
+	)
+)
+
+func init() {
+	prometheus.MustRegister(successfullScrapes)
+	prometheus.MustRegister(failedScrapes)
+	prometheus.MustRegister(timeseriesPushed)
+	prometheus.MustRegister(timeseriesDropped)
+}
+
+func observeSuccessfullScrape(source string) {
+	successfullScrapes.WithLabelValues(source).Inc()
+}
+
+func observeFailedScrape(source string) {
+	failedScrapes.WithLabelValues(source).Inc()
+}
+
+func observeSuccessfullRequest(batchSize int) {
+	timeseriesPushed.Add(float64(batchSize))
+}
+
+func observeFailedRequest(batchSize int) {
+	timeseriesDropped.Add(float64(batchSize))
+}

kubelet-to-gcm/monitor/poll.go

@@ -45,16 +45,18 @@ type MetricsSource interface {
 func Once(src MetricsSource, gcm *v3.Service) {
 	req, err := src.GetTimeSeriesReq()
 	if err != nil {
+		observeFailedScrape(src.Name())
 		log.Warningf("Failed to create time series request: %v", err)
 		return
 	}
+	observeSuccessfullScrape(src.Name())
 
 	for _, subReq := range subRequests(req) {
 		// Push that data to GCM's v3 API.
 		createCall := gcm.Projects.TimeSeries.Create(src.ProjectPath(), subReq)
 		if empty, err := createCall.Do(); err != nil {
 			log.Warningf("Failed to write time series data, empty: %v, err: %v", empty, err)
-
+			observeFailedRequest(len(subReq.TimeSeries))
 			jsonReq, err := subReq.MarshalJSON()
 			if err != nil {
 				log.Warningf("Failed to marshal time series as JSON")
@@ -64,6 +66,7 @@ func Once(src MetricsSource, gcm *v3.Service) {
 			return
 		}
 		log.V(4).Infof("Successfully wrote TimeSeries data for %s to GCM v3 API.", src.Name())
+		observeSuccessfullRequest(len(subReq.TimeSeries))
 	}
 }