add healthz to eventer

Author: andyxning

events/api/api.go

@@ -0,0 +1,28 @@
+// Copyright 2017 Google Inc. All Rights Reserved.
+//
+// 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 api
+
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apiserver/pkg/server/healthz"
+)
+
+func init() {
+	healthz.InstallHandler(http.DefaultServeMux, healthzChecker())
+
+	http.Handle("/metrics", prometheus.UninstrumentedHandler())
+}

events/api/handler.go

@@ -0,0 +1,47 @@
+// Copyright 2017 Google Inc. All Rights Reserved.
+//
+// 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 api
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/golang/glog"
+	"k8s.io/apiserver/pkg/server/healthz"
+	"k8s.io/heapster/events/manager"
+)
+
+const (
+	// MaxEventsScrapeDelay should be larger than `frequency` command line argument.
+	MaxEventsScrapeDelay = 3 * time.Minute
+)
+
+func healthzChecker() healthz.HealthzChecker {
+	return healthz.NamedCheck("healthz", func(r *http.Request) error {
+		if time.Since(manager.LatestScrapeTime) > MaxEventsScrapeDelay {
+			msg := fmt.Sprintf(
+				"No event batch within %s (latest: %s)",
+				MaxEventsScrapeDelay,
+				manager.LatestScrapeTime,
+			)
+			glog.Warning(msg)
+			return errors.New(msg)
+		}
+
+		return nil
+	})
+}

events/eventer.go

@@ -19,26 +19,32 @@ package main
 import (
 	"flag"
 	"fmt"
+	"net"
+	"net/http"
 	"os"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/golang/glog"
 	"k8s.io/apiserver/pkg/util/logs"
 	"k8s.io/heapster/common/flags"
+	"k8s.io/heapster/events/api"
 	"k8s.io/heapster/events/manager"
 	"k8s.io/heapster/events/sinks"
 	"k8s.io/heapster/events/sources"
 	"k8s.io/heapster/version"
 )
 
 var (
-	argFrequency = flag.Duration("frequency", 30*time.Second, "The frequency at which Eventer pushes events to sinks")
-	argMaxProcs  = flag.Int("max_procs", 0, "Max number of logical CPUs that can be used simultaneously, default to number of logical CPUs available on the machine when not specified or less than 1")
-	argSources   flags.Uris
-	argSinks     flags.Uris
-	argVersion   bool
+	argFrequency   = flag.Duration("frequency", 30*time.Second, "The resolution at which Eventer pushes events to sinks")
+	argMaxProcs    = flag.Int("max_procs", 0, "max number of CPUs that can be used simultaneously. Less than 1 for default (number of cores)")
+	argSources     flags.Uris
+	argSinks       flags.Uris
+	argVersion     bool
+	argHealthzIP   = flag.String("healthz-ip", "0.0.0.0", "ip eventer health check service uses")
+	argHealthzPort = flag.Uint("healthz-port", 8084, "port eventer health check listens on")
 )
 
 func main() {
@@ -98,16 +104,34 @@ func main() {
 	if err != nil {
 		glog.Fatalf("Failed to create main manager: %v", err)
 	}
-	manager.Start()
 
+	manager.Start()
 	glog.Infof("Starting eventer")
+
+	go startHTTPServer()
+
 	<-quitChannel
 }
 
+func startHTTPServer() {
+	glog.Info("Starting eventer http service")
+
+	glog.Fatal(http.ListenAndServe(net.JoinHostPort(*argHealthzIP, strconv.Itoa(int(*argHealthzPort))), nil))
+}
+
 func validateFlags() error {
-	if *argFrequency < 5*time.Second {
-		return fmt.Errorf("frequency needs to be greater than 5 seconds - %d", *argFrequency)
+	var minFrequency = 5 * time.Second
+
+	if *argFrequency < minFrequency {
+		return fmt.Errorf("frequency needs to be greater than %s, supplied %s", minFrequency,
+			*argFrequency)
 	}
+
+	if *argFrequency > api.MaxEventsScrapeDelay {
+		return fmt.Errorf("frequency needs to be smaller than %s, supplied %s",
+			api.MaxEventsScrapeDelay, *argFrequency)
+	}
+
 	return nil
 }
 

events/manager/manager.go

@@ -31,6 +31,9 @@ var (
 			Name:      "last_time_seconds",
 			Help:      "Last time of eventer housekeep since unix epoch in seconds.",
 		})
+
+	// Time of latest scrape operation
+	LatestScrapeTime = time.Now()
 )
 
 func init() {
@@ -89,6 +92,8 @@ func (rm *realManager) Housekeep() {
 func (rm *realManager) housekeep() {
 	defer lastHousekeepTimestamp.Set(float64(time.Now().Unix()))
 
+	LatestScrapeTime = time.Now()
+
 	// No parallelism. Assumes that the events are pushed to Heapster. Add parallelism
 	// when this stops to be true.
 	events := rm.source.GetNewEvents()

events/sources/kubernetes/kubernetes_source.go

@@ -96,6 +96,7 @@ event_loop:
 	}
 
 	totalEventsNum.Add(float64(len(result.Events)))
+
 	return &result
 }