Adding stackdriver exporter

pkg/exporters/register.go

@@ -0,0 +1,63 @@
+/*
+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 exporters
+
+import (
+	"fmt"
+
+	"k8s.io/node-problem-detector/pkg/types"
+)
+
+var (
+	handlers = make(map[types.ExporterType]types.ExporterHandler)
+)
+
+// Register registers a exporter factory method, which will be used to create the exporter.
+func Register(exporterType types.ExporterType, handler types.ExporterHandler) {
+	handlers[exporterType] = handler
+}
+
+// GetExporterNames retrieves all available exporter types.
+func GetExporterNames() []types.ExporterType {
+	exporterTypes := []types.ExporterType{}
+	for exporterType := range handlers {
+		exporterTypes = append(exporterTypes, exporterType)
+	}
+	return exporterTypes
+}
+
+// GetExporterHandlerOrDie retrieves the ExporterHandler for a specific type of exporter, panic if error occurs..
+func GetExporterHandlerOrDie(exporterType types.ExporterType) types.ExporterHandler {
+	handler, ok := handlers[exporterType]
+	if !ok {
+		panic(fmt.Sprintf("Exporter handler for %v does not exist", exporterType))
+	}
+	return handler
+}
+
+// NewExporters creates all exporters based on the configurations initialized.
+func NewExporters() []types.Exporter {
+	exporters := []types.Exporter{}
+	for _, handler := range handlers {
+		exporter := handler.CreateExporterOrDie(handler.Options)
+		if exporter == nil {
+			continue
+		}
+		exporters = append(exporters, exporter)
+	}
+	return exporters
+}

pkg/exporters/register_test.go

@@ -0,0 +1,69 @@
+/*
+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 exporters
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/node-problem-detector/pkg/types"
+)
+
+func TestRegistration(t *testing.T) {
+	fooExporterFactory := func(types.CommandLineOptions) types.Exporter {
+		return nil
+	}
+	fooExporterHandler := types.ExporterHandler{
+		CreateExporterOrDie: fooExporterFactory,
+		Options:             nil,
+	}
+
+	barExporterFactory := func(types.CommandLineOptions) types.Exporter {
+		return nil
+	}
+	barExporterHandler := types.ExporterHandler{
+		CreateExporterOrDie: barExporterFactory,
+		Options:             nil,
+	}
+
+	Register("foo", fooExporterHandler)
+	Register("bar", barExporterHandler)
+
+	expectedExporterNames := []types.ExporterType{"foo", "bar"}
+	exporterNames := GetExporterNames()
+	assert.ElementsMatch(t, expectedExporterNames, exporterNames)
+
+	handlers = make(map[types.ExporterType]types.ExporterHandler)
+}
+
+func TestGetExporterHandlerOrDie(t *testing.T) {
+	fooExporterFactory := func(types.CommandLineOptions) types.Exporter {
+		return nil
+	}
+	fooExporterHandler := types.ExporterHandler{
+		CreateExporterOrDie: fooExporterFactory,
+		Options:             nil,
+	}
+
+	Register("foo", fooExporterHandler)
+
+	assert.NotPanics(t, func() { GetExporterHandlerOrDie("foo") })
+	assert.Panics(t, func() { GetExporterHandlerOrDie("bar") })
+
+	handlers = make(map[types.ExporterType]types.ExporterHandler)
+}

pkg/exporters/stackdriver/config/config.go

@@ -0,0 +1,56 @@
+/*
+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 config
+
+import (
+	"time"
+
+	"k8s.io/node-problem-detector/pkg/exporters/stackdriver/gce"
+)
+
+var (
+	defaultExportPeriod          = (60 * time.Second).String()
+	defaultEndpoint              = "monitoring.googleapis.com:443"
+	defaultMetadataFetchTimeout  = (600 * time.Second).String()
+	defaultMetadataFetchInterval = (10 * time.Second).String()
+)
+
+type StackdriverExporterConfig struct {
+	ExportPeriod                string       `json:"exportPeriod"`
+	APIEndpoint                 string       `json:"apiEndpoint"`
+	GCEMetadata                 gce.Metadata `json:"gceMetadata"`
+	MetadataFetchTimeout        string       `json:"metadataFetchTimeout"`
+	MetadataFetchInterval       string       `json:"metadataFetchInterval"`
+	PanicOnMetadataFetchFailure bool         `json:"panicOnMetadataFetchFailure"`
+	CustomMetricPrefix          string       `json:"customMetricPrefix"`
+}
+
+// ApplyConfiguration applies default configurations.
+func (sec *StackdriverExporterConfig) ApplyConfiguration() {
+	if sec.ExportPeriod == "" {
+		sec.ExportPeriod = defaultExportPeriod
+	}
+	if sec.MetadataFetchTimeout == "" {
+		sec.MetadataFetchTimeout = defaultMetadataFetchTimeout
+	}
+	if sec.MetadataFetchInterval == "" {
+		sec.MetadataFetchInterval = defaultMetadataFetchInterval
+	}
+	if sec.APIEndpoint == "" {
+		sec.APIEndpoint = defaultEndpoint
+	}
+}

pkg/exporters/stackdriver/config/config_test.go

@@ -0,0 +1,107 @@
+/*
+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 config
+
+import (
+	"reflect"
+	"testing"
+
+	"k8s.io/node-problem-detector/pkg/exporters/stackdriver/gce"
+)
+
+func TestApplyConfiguration(t *testing.T) {
+	testCases := []struct {
+		name          string
+		orignalConfig StackdriverExporterConfig
+		wantedConfig  StackdriverExporterConfig
+	}{
+		{
+			name: "normal",
+			orignalConfig: StackdriverExporterConfig{
+				ExportPeriod:          "60s",
+				MetadataFetchTimeout:  "600s",
+				MetadataFetchInterval: "10s",
+				APIEndpoint:           "monitoring.googleapis.com:443",
+				GCEMetadata: gce.Metadata{
+					ProjectID:    "some-gcp-project",
+					Zone:         "us-central1-a",
+					InstanceID:   "56781234",
+					InstanceName: "some-gce-instance",
+				},
+			},
+			wantedConfig: StackdriverExporterConfig{
+				ExportPeriod:          "60s",
+				MetadataFetchTimeout:  "600s",
+				MetadataFetchInterval: "10s",
+				APIEndpoint:           defaultEndpoint,
+				GCEMetadata: gce.Metadata{
+					ProjectID:    "some-gcp-project",
+					Zone:         "us-central1-a",
+					InstanceID:   "56781234",
+					InstanceName: "some-gce-instance",
+				},
+			},
+		},
+		{
+			name: "staging API endpoint",
+			orignalConfig: StackdriverExporterConfig{
+				ExportPeriod:          "60s",
+				MetadataFetchTimeout:  "600s",
+				MetadataFetchInterval: "10s",
+				APIEndpoint:           "staging-monitoring.sandbox.googleapis.com:443",
+				GCEMetadata: gce.Metadata{
+					ProjectID:    "some-gcp-project",
+					Zone:         "us-central1-a",
+					InstanceID:   "56781234",
+					InstanceName: "some-gce-instance",
+				},
+			},
+			wantedConfig: StackdriverExporterConfig{
+				ExportPeriod:          "60s",
+				MetadataFetchTimeout:  "600s",
+				MetadataFetchInterval: "10s",
+				APIEndpoint:           "staging-monitoring.sandbox.googleapis.com:443",
+				GCEMetadata: gce.Metadata{
+					ProjectID:    "some-gcp-project",
+					Zone:         "us-central1-a",
+					InstanceID:   "56781234",
+					InstanceName: "some-gce-instance",
+				},
+			},
+		},
+		{
+			name:          "empty",
+			orignalConfig: StackdriverExporterConfig{},
+			wantedConfig: StackdriverExporterConfig{
+				ExportPeriod:          "1m0s",
+				MetadataFetchTimeout:  "10m0s",
+				MetadataFetchInterval: "10s",
+				APIEndpoint:           "monitoring.googleapis.com:443",
+				GCEMetadata:           gce.Metadata{},
+			},
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			test.orignalConfig.ApplyConfiguration()
+			if !reflect.DeepEqual(test.orignalConfig, test.wantedConfig) {
+				t.Errorf("Wanted: %+v. \nGot: %+v", test.wantedConfig, test.orignalConfig)
+			}
+		})
+	}
+}

pkg/exporters/stackdriver/gce/type.go

@@ -0,0 +1,67 @@
+/*
+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 gce
+
+import (
+	"cloud.google.com/go/compute/metadata"
+	"github.com/golang/glog"
+)
+
+type Metadata struct {
+	ProjectID    string `json:"projectID"`
+	Zone         string `json:"zone"`
+	InstanceID   string `json:"instanceID"`
+	InstanceName string `json:"instanceName"`
+}
+
+func (md *Metadata) HasMissingField() bool {
+	if md.ProjectID == "" || md.Zone == "" || md.InstanceID == "" || md.InstanceName == "" {
+		return true
+	}
+	return false
+}
+
+func (md *Metadata) PopulateFromGCE() error {
+	var err error
+	glog.Info("Fetching GCE metadata from metadata server")
+	if md.ProjectID == "" {
+		md.ProjectID, err = metadata.ProjectID()
+		if err != nil {
+			return err
+		}
+	}
+	if md.Zone == "" {
+		md.Zone, err = metadata.Zone()
+		if err != nil {
+			return err
+		}
+	}
+	if md.InstanceID == "" {
+		md.InstanceID, err = metadata.InstanceID()
+		if err != nil {
+			return err
+		}
+	}
+	if md.InstanceName == "" {
+		md.InstanceName, err = metadata.InstanceName()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

pkg/exporters/stackdriver/stackdriver_exporter.go

@@ -0,0 +1,190 @@
+/*
+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 stackdriverexporter
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"path/filepath"
+	"reflect"
+	"time"
+
+	"contrib.go.opencensus.io/exporter/stackdriver"
+	monitoredres "contrib.go.opencensus.io/exporter/stackdriver/monitoredresource"
+	"github.com/golang/glog"
+	"github.com/spf13/pflag"
+	"go.opencensus.io/stats/view"
+	"google.golang.org/api/option"
+
+	"github.com/avast/retry-go"
+	"k8s.io/node-problem-detector/pkg/exporters"
+	seconfig "k8s.io/node-problem-detector/pkg/exporters/stackdriver/config"
+	"k8s.io/node-problem-detector/pkg/types"
+	"k8s.io/node-problem-detector/pkg/util/metrics"
+)
+
+func init() {
+	clo := commandLineOptions{}
+	exporters.Register(exporterName, types.ExporterHandler{
+		CreateExporterOrDie: NewExporterOrDie,
+		Options:             &clo})
+}
+
+const exporterName = "stackdriver"
+
+var NPDMetricToSDMetric = map[metrics.MetricID]string{
+	metrics.HostUptimeID:      "compute.googleapis.com/guest/system/uptime",
+	metrics.ProblemCounterID:  "compute.googleapis.com/guest/system/problem_count",
+	metrics.DiskAvgQueueLenID: "compute.googleapis.com/guest/disk/queue_length",
+	metrics.DiskIOTimeID:      "compute.googleapis.com/guest/disk/io_time",
+	metrics.DiskWeightedIOID:  "compute.googleapis.com/guest/disk/weighted_io_time",
+}
+
+func getMetricTypeConversionFunction(customMetricPrefix string) func(*view.View) string {
+
+	return func(view *view.View) string {
+		viewName := view.Measure.Name()
+
+		fallbackMetricType := ""
+		if customMetricPrefix != "" {
+			// Example fallbackMetricType: custom.googleapis.com/npd/host/uptime
+			fallbackMetricType = filepath.Join(customMetricPrefix, viewName)
+		}
+
+		metricID, ok := metrics.MetricMap.ViewNameToMetricID(viewName)
+		if !ok {
+			return fallbackMetricType
+		}
+		stackdriverMetricType, ok := NPDMetricToSDMetric[metricID]
+		if !ok {
+			return fallbackMetricType
+		}
+		return stackdriverMetricType
+	}
+}
+
+type stackdriverExporter struct {
+	config seconfig.StackdriverExporterConfig
+}
+
+func (se *stackdriverExporter) setupOpenCensusViewExporterOrDie() {
+	clientOption := option.WithEndpoint(se.config.APIEndpoint)
+
+	var globalLabels stackdriver.Labels
+	globalLabels.Set("instance_name", se.config.GCEMetadata.InstanceName, "The name of the VM instance")
+
+	viewExporter, err := stackdriver.NewExporter(stackdriver.Options{
+		ProjectID:               se.config.GCEMetadata.ProjectID,
+		MonitoringClientOptions: []option.ClientOption{clientOption},
+		MonitoredResource: &monitoredres.GCEInstance{
+			ProjectID:  se.config.GCEMetadata.ProjectID,
+			InstanceID: se.config.GCEMetadata.InstanceID,
+			Zone:       se.config.GCEMetadata.Zone,
+		},
+		GetMetricType:           getMetricTypeConversionFunction(se.config.CustomMetricPrefix),
+		DefaultMonitoringLabels: &globalLabels,
+	})
+	if err != nil {
+		glog.Fatalf("Failed to create Stackdriver OpenCensus view exporter: %v", err)
+	}
+
+	exportPeriod, err := time.ParseDuration(se.config.ExportPeriod)
+	if err != nil {
+		glog.Fatalf("Failed to parse ExportPeriod %q: %v", se.config.ExportPeriod, err)
+	}
+
+	view.SetReportingPeriod(exportPeriod)
+	view.RegisterExporter(viewExporter)
+}
+
+func (se *stackdriverExporter) populateMetadataOrDie() {
+	if !se.config.GCEMetadata.HasMissingField() {
+		glog.Infof("Using GCE metadata specified in the config file: %+v", se.config.GCEMetadata)
+		return
+	}
+
+	metadataFetchTimeout, err := time.ParseDuration(se.config.MetadataFetchTimeout)
+	if err != nil {
+		glog.Fatalf("Failed to parse MetadataFetchTimeout %q: %v", se.config.MetadataFetchTimeout, err)
+	}
+
+	metadataFetchInterval, err := time.ParseDuration(se.config.MetadataFetchInterval)
+	if err != nil {
+		glog.Fatalf("Failed to parse MetadataFetchInterval %q: %v", se.config.MetadataFetchInterval, err)
+	}
+
+	glog.Infof("Populating GCE metadata by querying GCE metadata server.")
+	err = retry.Do(se.config.GCEMetadata.PopulateFromGCE,
+		retry.Delay(metadataFetchInterval),
+		retry.Attempts(uint(metadataFetchTimeout/metadataFetchInterval)),
+		retry.DelayType(retry.FixedDelay))
+	if err == nil {
+		glog.Infof("Using GCE metadata: %+v", se.config.GCEMetadata)
+		return
+	}
+	if se.config.PanicOnMetadataFetchFailure {
+		glog.Fatalf("Failed to populate GCE metadata: %v", err)
+	} else {
+		glog.Errorf("Failed to populate GCE metadata: %v", err)
+	}
+}
+
+// ExportProblems does nothing.
+// Stackdriver exporter only exports metrics.
+func (se *stackdriverExporter) ExportProblems(status *types.Status) {
+	return
+}
+
+type commandLineOptions struct {
+	configPath string
+}
+
+func (clo *commandLineOptions) SetFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&clo.configPath, "exporter.stackdriver", "",
+		"Configuration for Stackdriver exporter. Set to config file path.")
+}
+
+// NewExporterOrDie creates an exporter to export metrics to Stackdriver, panics if error occurs.
+func NewExporterOrDie(clo types.CommandLineOptions) types.Exporter {
+	options, ok := clo.(*commandLineOptions)
+	if !ok {
+		glog.Fatalf("Wrong type for the command line options of Stackdriver Exporter: %s.", reflect.TypeOf(clo))
+	}
+	if options.configPath == "" {
+		return nil
+	}
+
+	se := stackdriverExporter{}
+
+	// Apply configurations.
+	f, err := ioutil.ReadFile(options.configPath)
+	if err != nil {
+		glog.Fatalf("Failed to read configuration file %q: %v", options.configPath, err)
+	}
+	err = json.Unmarshal(f, &se.config)
+	if err != nil {
+		glog.Fatalf("Failed to unmarshal configuration file %q: %v", options.configPath, err)
+	}
+	se.config.ApplyConfiguration()
+
+	glog.Infof("Starting Stackdriver exporter %s", options.configPath)
+
+	se.populateMetadataOrDie()
+	se.setupOpenCensusViewExporterOrDie()
+
+	return &se
+}

pkg/exporters/stackdriver/stackdriver_exporter_test.go

@@ -0,0 +1,33 @@
+// +build !disable_stackdriver_exporter
+
+/*
+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 stackdriverexporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/node-problem-detector/pkg/exporters"
+)
+
+func TestRegistration(t *testing.T) {
+	assert.NotPanics(t,
+		func() { exporters.GetExporterHandlerOrDie(exporterName) },
+		"Stackdriver exporter failed to register itself as an exporter.")
+}

pkg/types/types.go

@@ -18,6 +18,8 @@ package types
 
 import (
 	"time"
+
+	"github.com/spf13/pflag"
 )
 
 // The following types are used internally in problem detector. In the future this could be the
@@ -135,15 +137,14 @@ type ProblemDaemonHandler struct {
 // ExporterType is the type of the exporter.
 type ExporterType string
 
-// ExporterConfigPathMap represents configurations on all types of exporters:
-// 1) Each key represents a type of exporter.
-// 2) Each value represents the config file path for the exporter.
-type ExporterConfigPathMap map[ExporterType]*string
-
 // ExporterHandler represents the initialization handler for a type of exporter.
 type ExporterHandler struct {
 	// CreateExporterOrDie initializes an exporter, panic if error occurs.
-	CreateExporterOrDie func(string) Exporter
+	CreateExporterOrDie func(CommandLineOptions) Exporter
 	// CmdOptionDescription explains how to configure the exporter from command line arguments.
-	CmdOptionDescription string
+	Options CommandLineOptions
+}
+
+type CommandLineOptions interface {
+	SetFlags(*pflag.FlagSet)
 }