Merge pull request #28431 from neisw/trt-1042-data-collection

trt-1042: collect watch request counts

Author: openshift-merge-bot[bot]

pkg/dataloader/types.go

@@ -0,0 +1,64 @@
+package dataloader
+
+import (
+	"encoding/json"
+	"io/ioutil"
+)
+
+type DataType = string
+
+const (
+	DataTypeFloat64 DataType = "float64"
+	DataTypeString  DataType = "string"
+	DataTypeInteger DataType = "int64"
+	// RFC3339  based value "2006-01-02T15:04:05Z07:00
+	DataTypeTimestamp DataType = "timestamp"
+	DataTypeJSON      DataType = "json"
+
+	// files that end with this suffix will be automatically written to the specified table name via ci-data-loader
+	AutoDataLoaderSuffix = "autodl.json"
+)
+
+type DataFile struct {
+	// Table name to be created / updated with the corresponding data
+	TableName string `json:"table_name"`
+	// Schema identifying the data types associated with the row values
+	// JobRunName, PartitionTime and Source will be provided by default and do not need to be specified here
+	// Schema defined here are optional columns, unless used as PartitionColumn
+	// New columns will be added but columns that get removed here will *not* be deleted
+	// from the table in order to preserve integrity across releases
+	// However as optional columns the data does not have to be
+	// included if no longer necessary
+	// if breaking changes are needed best to define a new table name
+	Schema map[string]DataType `json:"schema"`
+	// If the existing row key differs from the specified schema column name you need to map a row key to a different schema name rowKey->newName
+	SchemaMapping map[string]string `json:"schema_mapping"`
+	// The data to be uploaded
+	Rows []map[string]string `json:"rows"`
+
+	// Optional
+	// Depending on the size of your data the rows might have to be chunked
+	// when writing.  Default chunk size is 5k rows.
+	// If the row data is large this can be changed to make smaller chunks
+	ChunkSize int `json:"chunk_size"`
+
+	// ExpirationDays and PartitionColumn will only
+	// be used when first creating the table
+	// if the table exists changing these
+	// values will not update the table
+	// Default expiration days is 365
+	ExpirationDays int `json:"expiration_days"`
+	// A partition column, PartitionTime, will automatically be added with the value
+	// of the file creation timestamp.  If your data has a timestamp value already
+	// it can be specified as the partition column instead
+	// and the default PartitionTime will be omitted
+	PartitionColumn string `json:"partition_column"`
+}
+
+func WriteDataFile(filename string, dataFile DataFile) error {
+	jsonContent, err := json.MarshalIndent(dataFile, "", "    ")
+	if err != nil {
+		return err
+	}
+	return ioutil.WriteFile(filename, jsonContent, 0644)
+}

pkg/defaultmonitortests/types.go

@@ -2,7 +2,7 @@ package defaultmonitortests
 
 import (
 	"fmt"
-
+	"github.com/openshift/origin/pkg/monitortests/testframework/watchrequestcountscollector"
 	"github.com/sirupsen/logrus"
 
 	"github.com/openshift/origin/pkg/monitortests/kubeapiserver/disruptionnewapiserver"
@@ -168,6 +168,7 @@ func newUniversalMonitorTests() monitortestframework.MonitorTestRegistry {
 	monitorTestRegistry.AddMonitorTestOrDie("clusteroperator-collector", "Test Framework", watchclusteroperators.NewOperatorWatcher())
 
 	monitorTestRegistry.AddMonitorTestOrDie("azure-metrics-collector", "Test Framework", azuremetrics.NewAzureMetricsCollector())
+	monitorTestRegistry.AddMonitorTestOrDie("watch-request-counts-collector", "Test Framework", watchrequestcountscollector.NewWatchRequestCountSerializer())
 
 	return monitorTestRegistry
 }

pkg/monitortests/testframework/watchrequestcountscollector/monitortest.go

@@ -0,0 +1,189 @@
+package watchrequestcountscollector
+
+import (
+	"context"
+	"fmt"
+	apiserverclientv1 "github.com/openshift/client-go/apiserver/clientset/versioned/typed/apiserver/v1"
+	"github.com/openshift/origin/pkg/dataloader"
+	"github.com/openshift/origin/pkg/monitor/monitorapi"
+	"github.com/openshift/origin/pkg/monitortestframework"
+	"github.com/openshift/origin/pkg/test/ginkgo/junitapi"
+	exutil "github.com/openshift/origin/test/extended/util"
+	"github.com/sirupsen/logrus"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/rest"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type watchRequestCountSerializer struct {
+	monitorStartTime time.Time
+	adminRESTConfig  *rest.Config
+}
+
+func NewWatchRequestCountSerializer() monitortestframework.MonitorTest {
+	return &watchRequestCountSerializer{}
+}
+
+func (w *watchRequestCountSerializer) StartCollection(ctx context.Context, adminRESTConfig *rest.Config, recorder monitorapi.RecorderWriter) error {
+	w.monitorStartTime = time.Now()
+	w.adminRESTConfig = adminRESTConfig
+	return nil
+}
+
+func (w *watchRequestCountSerializer) CollectData(ctx context.Context, storageDir string, beginning, end time.Time) (monitorapi.Intervals, []*junitapi.JUnitTestCase, error) {
+	return nil, nil, nil
+}
+
+func (w *watchRequestCountSerializer) ConstructComputedIntervals(ctx context.Context, startingIntervals monitorapi.Intervals, recordedResources monitorapi.ResourcesMap, beginning, end time.Time) (constructedIntervals monitorapi.Intervals, err error) {
+	return nil, nil
+}
+
+func (w *watchRequestCountSerializer) EvaluateTestsFromConstructedIntervals(ctx context.Context, finalIntervals monitorapi.Intervals) ([]*junitapi.JUnitTestCase, error) {
+	return nil, nil
+}
+
+func (w *watchRequestCountSerializer) WriteContentToStorage(ctx context.Context, storageDir, timeSuffix string, finalIntervals monitorapi.Intervals, finalResourceState monitorapi.ResourcesMap) error {
+	oc := exutil.NewCLIWithoutNamespace("api-requests")
+
+	infra, err := oc.AdminConfigClient().ConfigV1().Infrastructures().Get(context.Background(), "cluster", metav1.GetOptions{})
+	if err != nil {
+		logrus.WithError(err).Warn("unable to get cluster infrastructure")
+		return nil
+	}
+
+	watchRequestCounts, err := GetWatchRequestCounts(ctx, oc)
+	if err != nil {
+		logrus.WithError(err).Warn("unable to get watch request counts")
+		return nil
+	}
+
+	// infra.Status.ControlPlaneTopology, infra.Spec.PlatformSpec.Type, operator, value
+	rows := make([]map[string]string, 0)
+	for _, item := range watchRequestCounts {
+		operator := strings.Split(item.Operator, ":")[3]
+		rows = append(rows, map[string]string{"ControlPlaneTopology": string(infra.Status.ControlPlaneTopology), "PlatformType": string(infra.Spec.PlatformSpec.Type), "Operator": operator, "WatchRequestCount": strconv.FormatInt(item.Count, 10)})
+	}
+
+	dataFile := dataloader.DataFile{
+		TableName: "operator_watch_requests",
+		Schema:    map[string]dataloader.DataType{"ControlPlaneTopology": dataloader.DataTypeString, "PlatformType": dataloader.DataTypeString, "Operator": dataloader.DataTypeString, "WatchRequestCount": dataloader.DataTypeInteger},
+		Rows:      rows,
+	}
+	fileName := filepath.Join(storageDir, fmt.Sprintf("operator-watch-requests%s-%s", timeSuffix, dataloader.AutoDataLoaderSuffix))
+	err = dataloader.WriteDataFile(fileName, dataFile)
+	if err != nil {
+		logrus.WithError(err).Warnf("unable to write data file: %s", fileName)
+		return nil
+	}
+
+	return nil
+}
+
+func (w *watchRequestCountSerializer) Cleanup(ctx context.Context) error {
+	return nil
+}
+
+type OperatorKey struct {
+	NodeName string
+	Operator string
+	Hour     int
+}
+
+type RequestCount struct {
+	NodeName string
+	Operator string
+	Count    int64
+	Hour     int
+}
+
+func GetWatchRequestCounts(ctx context.Context, oc *exutil.CLI) ([]*RequestCount, error) {
+
+	apirequestCountClient, err := apiserverclientv1.NewForConfig(oc.AdminConfig())
+	if err != nil {
+		logrus.WithError(err).Warn("unable to initialize apirequestCountClient")
+		return nil, err
+	}
+
+	apiRequestCounts, err := apirequestCountClient.APIRequestCounts().List(ctx, metav1.ListOptions{})
+
+	watchRequestCounts := []*RequestCount{}
+	watchRequestCountsMap := map[OperatorKey]*RequestCount{}
+
+	for _, apiRequestCount := range apiRequestCounts.Items {
+		if apiRequestCount.Status.RequestCount <= 0 {
+			continue
+		}
+		for hourIdx, perResourceAPIRequestLog := range apiRequestCount.Status.Last24h {
+			if perResourceAPIRequestLog.RequestCount > 0 {
+				for _, perNodeCount := range perResourceAPIRequestLog.ByNode {
+					if perNodeCount.RequestCount <= 0 {
+						continue
+					}
+					for _, perUserCount := range perNodeCount.ByUser {
+						if perUserCount.RequestCount <= 0 {
+							continue
+						}
+						// take only operators into account
+						if !strings.HasSuffix(perUserCount.UserName, "-operator") {
+							continue
+						}
+						for _, verb := range perUserCount.ByVerb {
+							if verb.Verb != "watch" || verb.RequestCount == 0 {
+								continue
+							}
+							key := OperatorKey{
+								NodeName: perNodeCount.NodeName,
+								Operator: perUserCount.UserName,
+								Hour:     hourIdx,
+							}
+							// group requests by a resource (the number of watchers in the code does not change
+							// so much as the number of requests)
+							if _, exists := watchRequestCountsMap[key]; exists {
+								watchRequestCountsMap[key].Count += verb.RequestCount
+							} else {
+								watchRequestCountsMap[key] = &RequestCount{
+									NodeName: perNodeCount.NodeName,
+									Operator: perUserCount.UserName,
+									Count:    verb.RequestCount,
+									Hour:     hourIdx,
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+
+	// take maximum from all hours through all nodes
+	watchRequestCountsMapMax := map[OperatorKey]*RequestCount{}
+	for _, requestCount := range watchRequestCountsMap {
+		key := OperatorKey{
+			Operator: requestCount.Operator,
+		}
+		if _, exists := watchRequestCountsMapMax[key]; exists {
+			if watchRequestCountsMapMax[key].Count < requestCount.Count {
+				watchRequestCountsMapMax[key].Count = requestCount.Count
+				watchRequestCountsMapMax[key].NodeName = requestCount.NodeName
+				watchRequestCountsMapMax[key].Hour = requestCount.Hour
+			}
+		} else {
+			watchRequestCountsMapMax[key] = requestCount
+		}
+	}
+
+	// sort the requsts counts so it's easy to see the biggest offenders
+	for _, requestCount := range watchRequestCountsMapMax {
+		watchRequestCounts = append(watchRequestCounts, requestCount)
+	}
+
+	sort.Slice(watchRequestCounts, func(i int, j int) bool {
+		return watchRequestCounts[i].Count > watchRequestCounts[j].Count
+	})
+
+	return watchRequestCounts, nil
+}