ApiRequestCounts conditional gathering

Author: tremes

manifests/03-clusterrole.yaml

@@ -212,6 +212,14 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - apiserver.openshift.io
+    resources:
+      - apirequestcounts
+    verbs:
+      - get
+      - list
+      - watch  
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding

pkg/gatherers/conditional/conditional_gatherer.go

@@ -27,6 +27,7 @@ import (
 var gatheringFunctionBuilders = map[GatheringFunctionName]GathererFunctionBuilderPtr{
 	GatherLogsOfNamespace:         (*Gatherer).BuildGatherLogsOfNamespace,
 	GatherImageStreamsOfNamespace: (*Gatherer).BuildGatherImageStreamsOfNamespace,
+	GatherAPIRequestCounts:        (*Gatherer).BuildGatherAPIRequestCounts,
 }
 
 // gatheringRules contains all the rules used to run conditional gatherings.
@@ -77,6 +78,21 @@ var defaultGatheringRules = []GatheringRule{
 			},
 		},
 	},
+	{
+		Conditions: []ConditionWithParams{
+			{
+				Type: AlertIsFiring,
+				Params: AlertIsFiringConditionParams{
+					Name: "APIRemovedInNextReleaseInUse",
+				},
+			},
+		},
+		GatheringFunctions: GatheringFunctions{
+			GatherAPIRequestCounts: GatherAPIRequestCountsParams{
+				AlertName: "APIRemovedInNextReleaseInUse",
+			},
+		},
+	},
 }
 
 const canConditionalGathererFail = false
@@ -86,8 +102,9 @@ type Gatherer struct {
 	gatherProtoKubeConfig   *rest.Config
 	metricsGatherKubeConfig *rest.Config
 	imageKubeConfig         *rest.Config
-	firingAlerts            map[string]bool // golang doesn't have sets :(
-	gatheringRules          []GatheringRule
+	// there can be multiple instances of the same alert
+	firingAlerts   map[string][]Alert // golang doesn't have sets :(
+	gatheringRules []GatheringRule
 }
 
 // New creates a new instance of conditional gatherer with the appropriate configs
@@ -138,7 +155,6 @@ func (g *Gatherer) GetGatheringFunctions(ctx context.Context) (map[string]gather
 		if err != nil {
 			return nil, err
 		}
-
 		if allConditionsAreSatisfied {
 			functions, errs := g.createGatheringClosures(conditionalGathering.GatheringFunctions)
 			if len(errs) > 0 {
@@ -206,7 +222,7 @@ func (g *Gatherer) updateAlertsCache(ctx context.Context) error {
 func (g *Gatherer) updateAlertsCacheFromClient(ctx context.Context, metricsClient rest.Interface) error {
 	const logPrefix = "conditional gatherer: "
 
-	g.firingAlerts = make(map[string]bool)
+	g.firingAlerts = make(map[string][]Alert)
 
 	data, err := metricsClient.Get().AbsPath("federate").
 		Param("match[]", `ALERTS{alertstate="firing"}`).
@@ -237,17 +253,31 @@ func (g *Gatherer) updateAlertsCacheFromClient(ctx context.Context, metricsClien
 			klog.Info(logPrefix + "metric is nil")
 			continue
 		}
-
+		var alert Alert
 		for _, label := range metric.GetLabel() {
 			if label == nil {
 				klog.Info(logPrefix + "label is nil")
 				continue
 			}
-
-			if label.GetName() == "alertname" {
-				g.firingAlerts[label.GetValue()] = true
+			switch label.GetName() {
+			case "alertname":
+				alert.Name = label.GetValue()
+			case "version":
+				alert.Version = label.GetValue()
+			case "resource":
+				alert.Resource = label.GetValue()
+			case "group":
+				alert.Group = label.GetValue()
 			}
 		}
+		alerts, ok := g.firingAlerts[alert.Name]
+		if ok {
+			alerts = append(alerts, alert)
+		} else {
+			alerts = []Alert{alert}
+		}
+
+		g.firingAlerts[alert.Name] = alerts
 	}
 
 	return nil

pkg/gatherers/conditional/conditional_gatherer_test.go

@@ -166,7 +166,7 @@ func Test_Gatherer_GatherConditionalGathererRules(t *testing.T) {
 	err = json.Unmarshal(item, &gotGatheringRules)
 	assert.NoError(t, err)
 
-	assert.Len(t, gotGatheringRules, 1)
+	assert.Len(t, gotGatheringRules, 2)
 }
 
 func newFakeClientWithMetrics(metrics string) *fake.RESTClient {

pkg/gatherers/conditional/gather_api_request_count.go

@@ -0,0 +1,107 @@
+package conditional
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/openshift/insights-operator/pkg/gatherers"
+	"github.com/openshift/insights-operator/pkg/record"
+	"github.com/openshift/insights-operator/pkg/utils"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+)
+
+// APIRequestCount defines a type used when marshaling into JSON
+type APIRequestCount struct {
+	ResourceName        string `json:"resource"`
+	RemovedInRelease    string `json:"removed_in_release"`
+	TotalRequestCount   int64  `json:"total_request_count"`
+	LastDayRequestCount int64  `json:"last_day_request_count"`
+}
+
+// BuildGatherApiRequestCounts creates a gathering closure which collects logs api requests counts for the
+// resources mentioned in
+// Params is of type AlertIsFiringConditionParams:
+//   - alert_name string - name of the firing alert
+//
+//
+// * Location in archive: conditional/api_request_counts.json
+// * Since versions:
+//   * 4.10+
+func (g *Gatherer) BuildGatherAPIRequestCounts(paramsInterface interface{}) (gatherers.GatheringClosure, error) {
+	params, ok := paramsInterface.(GatherAPIRequestCountsParams)
+	if !ok {
+		return gatherers.GatheringClosure{}, fmt.Errorf(
+			"unexpected type in paramsInterface, expected %T, got %T",
+			GatherAPIRequestCountsParams{}, paramsInterface,
+		)
+	}
+
+	return gatherers.GatheringClosure{
+		Run: func(ctx context.Context) ([]record.Record, []error) {
+			records, err := g.gatherAPIRequestCounts(ctx, params.AlertName)
+			if err != nil {
+				return records, err
+			}
+			return records, nil
+		},
+		CanFail: canConditionalGathererFail,
+	}, nil
+}
+
+func (g *Gatherer) gatherAPIRequestCounts(ctx context.Context, alertName string) ([]record.Record, []error) {
+	var resources []string
+	for _, a := range g.firingAlerts[alertName] {
+		resourceName := fmt.Sprintf("%s.%s.%s", a.Resource, a.Version, a.Group)
+		resources = append(resources, resourceName)
+	}
+
+	dynamicClient, err := dynamic.NewForConfig(g.gatherProtoKubeConfig)
+	if err != nil {
+		return nil, []error{err}
+	}
+	gvr := schema.GroupVersionResource{Group: "apiserver.openshift.io", Version: "v1", Resource: "apirequestcounts"}
+	apiReqCountsList, err := dynamicClient.Resource(gvr).List(ctx, metav1.ListOptions{})
+	if errors.IsNotFound(err) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, []error{err}
+	}
+	var records []record.Record
+	var errrs []error
+	var apiReqCounts []APIRequestCount
+	for i := range apiReqCountsList.Items {
+		it := apiReqCountsList.Items[i]
+
+		// filter only resources we're interested in
+		if utils.IsInSlice(it.GetName(), resources) {
+			totalReqCount, err := utils.NestedInt64Wrapper(it.Object, "status", "requestCount")
+			if err != nil {
+				errrs = append(errrs, err)
+			}
+			lastDayReqCount, err := utils.NestedInt64Wrapper(it.Object, "status", "currentHour", "requestCount")
+			if err != nil {
+				errrs = append(errrs, err)
+			}
+			removedInRel, err := utils.NestedStringWrapper(it.Object, "status", "removedInRelease")
+			if err != nil {
+				errrs = append(errrs, err)
+			}
+			apiReqCount := APIRequestCount{
+				TotalRequestCount:   totalReqCount,
+				LastDayRequestCount: lastDayReqCount,
+				ResourceName:        it.GetName(),
+				RemovedInRelease:    removedInRel,
+			}
+			apiReqCounts = append(apiReqCounts, apiReqCount)
+		}
+	}
+	records = append(records, record.Record{
+		Name: fmt.Sprintf("%v/api_request_counts", g.GetName()),
+		Item: record.JSONMarshaller{Object: apiReqCounts},
+	})
+	return records, errrs
+}

pkg/gatherers/conditional/gathering_functions.go

@@ -13,13 +13,19 @@ type GatheringFunctions = map[GatheringFunctionName]interface{}
 // GatheringFunctionName defines functions of conditional gatherer
 type GatheringFunctionName string
 
-// GatherLogsOfNamespace is a function collecting logs of the provided namespace.
-// See file gather_logs_of_namespace.go
-const GatherLogsOfNamespace GatheringFunctionName = "logs_of_namespace"
+const (
+	// GatherLogsOfNamespace is a function collecting logs of the provided namespace.
+	// See file gather_logs_of_namespace.go
+	GatherLogsOfNamespace GatheringFunctionName = "logs_of_namespace"
 
-// GatherImageStreamsOfNamespace is a function collecting image streams of the provided namespace.
-// See file gather_image_streams_of_namespace.go
-const GatherImageStreamsOfNamespace GatheringFunctionName = "image_streams_of_namespace"
+	// GatherImageStreamsOfNamespace is a function collecting image streams of the provided namespace.
+	// See file gather_image_streams_of_namespace.go
+	GatherImageStreamsOfNamespace GatheringFunctionName = "image_streams_of_namespace"
+
+	// GatherAPIRequestCounts is a function collecting api request counts for the resources read
+	// from the corresponding alert
+	GatherAPIRequestCounts GatheringFunctionName = "api_request_counts"
+)
 
 func (name GatheringFunctionName) NewParams(jsonParams []byte) (interface{}, error) {
 	switch name {
@@ -31,6 +37,10 @@ func (name GatheringFunctionName) NewParams(jsonParams []byte) (interface{}, err
 		var result GatherImageStreamsOfNamespaceParams
 		err := json.Unmarshal(jsonParams, &result)
 		return result, err
+	case GatherAPIRequestCounts:
+		var params GatherAPIRequestCountsParams
+		err := json.Unmarshal(jsonParams, &params)
+		return params, err
 	}
 	return nil, fmt.Errorf("unable to create params for %T: %v", name, name)
 }
@@ -50,3 +60,8 @@ type GatherImageStreamsOfNamespaceParams struct {
 	// Namespace from which to collect image streams
 	Namespace string `json:"namespace"`
 }
+
+// GatherAPIRequestCountsParams defines parameters for api_request_counts gatherer
+type GatherAPIRequestCountsParams struct {
+	AlertName string `json:"alert_name"`
+}

pkg/gatherers/conditional/gathering_rule.schema.json

@@ -120,6 +120,19 @@
                             "pattern": "^openshift-[a-zA-Z0-9_.-]{1,128}$"
                         }
                     }
+                },
+                "^api_request_counts$": {
+                    "type": "object",
+                    "title": "GatherApiRequestCountsParams",
+                    "required": [
+                        "alert_name"
+                    ],
+                    "properties": {
+                        "alert_name": {
+                            "type": "string",
+                            "title": "AlertName"
+                        }
+                    }
                 }
             }
         }

pkg/gatherers/conditional/types.go

@@ -15,3 +15,11 @@ type GatheringRule struct {
 
 // GathererFunctionBuilderPtr defines a pointer to a gatherer function builder
 type GathererFunctionBuilderPtr = func(*Gatherer, interface{}) (gatherers.GatheringClosure, error)
+
+// Alert defines basic alert attributes (basically alert labels)
+type Alert struct {
+	Name     string
+	Resource string
+	Version  string
+	Group    string
+}

pkg/utils/slices.go

@@ -40,3 +40,13 @@ func TakeLastNItemsFromByteArray(array []byte, desiredLength int) []byte {
 
 	return array
 }
+
+// IsInSlice checks if string is in the slice of strings
+func IsInSlice(v string, a []string) bool {
+	for _, s := range a {
+		if s == v {
+			return true
+		}
+	}
+	return false
+}

pkg/utils/unstructerd_wrappers.go

@@ -31,6 +31,18 @@ func NestedSliceWrapper(obj map[string]interface{}, fields ...string) ([]interfa
 	return s, nil
 }
 
+func NestedInt64Wrapper(obj map[string]interface{}, fields ...string) (int64, error) {
+	i, ok, err := unstructured.NestedInt64(obj, fields...)
+	if !ok {
+		return 0, fmt.Errorf("can't find %s", formatSlice(fields...))
+	}
+	if err != nil {
+		return 0, err
+	}
+
+	return i, nil
+}
+
 func formatSlice(s ...string) string {
 	var str string
 	for _, f := range s {