Move apiserver-reporting logic into k8s_exporter.

Added CLI option "enable-k8s-exporter" (default to true). Users can use
this option to enable/disable exporting to Kubernetes control plane.

This commit also removes all the apiserver-specific logic from package
problemdetector.

Future exporters (e.g. to local journald, Prometheus, other control
planes) should implement types.Exporter interface.

pkg/exporters/k8sexporter/condition/manager.go

@@ -21,7 +21,7 @@ import (
 	"sync"
 	"time"
 
-	"k8s.io/node-problem-detector/pkg/problemclient"
+	"k8s.io/node-problem-detector/pkg/exporters/k8sexporter/problemclient"
 	"k8s.io/node-problem-detector/pkg/types"
 	problemutil "k8s.io/node-problem-detector/pkg/util"
 

pkg/exporters/k8sexporter/condition/manager_test.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"k8s.io/node-problem-detector/pkg/problemclient"
+	"k8s.io/node-problem-detector/pkg/exporters/k8sexporter/problemclient"
 	"k8s.io/node-problem-detector/pkg/types"
 	problemutil "k8s.io/node-problem-detector/pkg/util"
 

pkg/exporters/k8sexporter/k8s_exporter.go

@@ -0,0 +1,93 @@
+/*
+Copyright 2019 The Kubernetes Authors 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 k8sexporter
+
+import (
+	"net"
+	"net/http"
+	_ "net/http/pprof"
+	"strconv"
+
+	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/util/clock"
+
+	"k8s.io/node-problem-detector/cmd/options"
+	"k8s.io/node-problem-detector/pkg/exporters/k8sexporter/condition"
+	"k8s.io/node-problem-detector/pkg/exporters/k8sexporter/problemclient"
+	"k8s.io/node-problem-detector/pkg/types"
+	"k8s.io/node-problem-detector/pkg/util"
+)
+
+type k8sExporter struct {
+	client           problemclient.Client
+	conditionManager condition.ConditionManager
+}
+
+// NewExporterOrDie creates a exporter for Kubernetes apiserver exporting, panics if error occurs.
+func NewExporterOrDie(npdo *options.NodeProblemDetectorOptions) types.Exporter {
+	if !npdo.EnableK8sExporter {
+		return nil
+	}
+
+	c := problemclient.NewClientOrDie(npdo)
+	ke := k8sExporter{
+		client:           c,
+		conditionManager: condition.NewConditionManager(c, clock.RealClock{}),
+	}
+
+	ke.startHTTPReporting(npdo)
+	ke.conditionManager.Start()
+
+	return &ke
+}
+
+func (ke *k8sExporter) ExportProblems(status *types.Status) {
+	for _, event := range status.Events {
+		ke.client.Eventf(util.ConvertToAPIEventType(event.Severity), status.Source, event.Reason, event.Message)
+	}
+	for _, cdt := range status.Conditions {
+		ke.conditionManager.UpdateCondition(cdt)
+	}
+}
+
+func (ke *k8sExporter) startHTTPReporting(npdo *options.NodeProblemDetectorOptions) {
+	if npdo.ServerPort <= 0 {
+		return
+	}
+	mux := http.NewServeMux()
+
+	// Add healthz http request handler. Always return ok now, add more health check
+	// logic in the future.
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("ok"))
+	})
+
+	// Add the handler to serve condition http request.
+	mux.HandleFunc("/conditions", func(w http.ResponseWriter, r *http.Request) {
+		util.ReturnHTTPJson(w, ke.conditionManager.GetConditions())
+	})
+
+	addr := net.JoinHostPort(npdo.ServerAddress, strconv.Itoa(npdo.ServerPort))
+	go func() {
+		err := http.ListenAndServe(addr, mux)
+		if err != nil {
+			glog.Fatalf("Failed to start server: %v", err)
+		}
+	}()
+}

pkg/problemdetector/problem_detector.go

@@ -18,43 +18,33 @@ package problemdetector
 
 import (
 	"fmt"
-	"net/http"
 
 	"github.com/golang/glog"
 
-	"k8s.io/apimachinery/pkg/util/clock"
-
-	"k8s.io/node-problem-detector/pkg/condition"
-	"k8s.io/node-problem-detector/pkg/problemclient"
 	"k8s.io/node-problem-detector/pkg/types"
-	"k8s.io/node-problem-detector/pkg/util"
 )
 
 // ProblemDetector collects statuses from all problem daemons and update the node condition and send node event.
 type ProblemDetector interface {
 	Run() error
-	RegisterHTTPHandlers()
 }
 
 type problemDetector struct {
-	client           problemclient.Client
-	conditionManager condition.ConditionManager
-	monitors         map[string]types.Monitor
+	monitors  map[string]types.Monitor
+	exporters []types.Exporter
 }
 
 // NewProblemDetector creates the problem detector. Currently we just directly passed in the problem daemons, but
 // in the future we may want to let the problem daemons register themselves.
-func NewProblemDetector(monitors map[string]types.Monitor, client problemclient.Client) ProblemDetector {
+func NewProblemDetector(monitors map[string]types.Monitor, exporters []types.Exporter) ProblemDetector {
 	return &problemDetector{
-		client:           client,
-		conditionManager: condition.NewConditionManager(client, clock.RealClock{}),
-		monitors:         monitors,
+		monitors:  monitors,
+		exporters: exporters,
 	}
 }
 
 // Run starts the problem detector.
 func (p *problemDetector) Run() error {
-	p.conditionManager.Start()
 	// Start the log monitors one by one.
 	var chans []<-chan *types.Status
 	for cfg, m := range p.monitors {
@@ -75,24 +65,13 @@ func (p *problemDetector) Run() error {
 	for {
 		select {
 		case status := <-ch:
-			for _, event := range status.Events {
-				p.client.Eventf(util.ConvertToAPIEventType(event.Severity), status.Source, event.Reason, event.Message)
-			}
-			for _, cdt := range status.Conditions {
-				p.conditionManager.UpdateCondition(cdt)
+			for _, exporter := range p.exporters {
+				exporter.ExportProblems(status)
 			}
 		}
 	}
 }
 
-// RegisterHTTPHandlers registers http handlers of node problem detector.
-func (p *problemDetector) RegisterHTTPHandlers() {
-	// Add the handler to serve condition http request.
-	http.HandleFunc("/conditions", func(w http.ResponseWriter, r *http.Request) {
-		util.ReturnHTTPJson(w, p.conditionManager.GetConditions())
-	})
-}
-
 func groupChannel(chans []<-chan *types.Status) <-chan *types.Status {
 	statuses := make(chan *types.Status)
 	for _, ch := range chans {

pkg/types/types.go

@@ -107,3 +107,9 @@ type Monitor interface {
 	// Stop stops the log monitor.
 	Stop()
 }
+
+// Exporter exports machine health data to certain control plane.
+type Exporter interface {
+	// Export problems to the control plane.
+	ExportProblems(*Status)
+}