Introduce insightsoperators.openshift.io CR & implement its gather status attribute

Author: tremes

manifests/03-clusterrole.yaml

@@ -59,6 +59,14 @@ rules:
       - clusteroperators
     verbs:
       - create
+  - apiGroups:
+      - "operator.openshift.io"
+    resources:
+      - insightsoperators
+    verbs:
+      - get
+      - update
+      - patch
   - apiGroups:
       - "config.openshift.io"
     resources:

manifests/03-prometheus_rolebinding.yaml manifests/04-prometheus_rolebinding.yaml

@@ -0,0 +1,13 @@
+apiVersion: operator.openshift.io/v1
+kind: InsightsOperator
+metadata:
+  name: cluster
+  annotations:
+    include.release.openshift.io/ibm-cloud-managed: "true"
+    include.release.openshift.io/self-managed-high-availability: "true"
+    include.release.openshift.io/single-node-developer: "true"
+    capability.openshift.io/name: Insights
+spec:
+  logLevel: Normal
+  managementState: Managed
+  operatorLogLevel: Normal

manifests/06-deployment-ibm-cloud-managed.yaml manifests/05-deployment-ibm-cloud-managed.yaml

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
+	operatorv1client "github.com/openshift/client-go/operator/clientset/versioned/typed/operator/v1"
 	"github.com/openshift/library-go/pkg/controller/controllercmd"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -63,6 +64,11 @@ func (s *Operator) Run(ctx context.Context, controller *controllercmd.Controller
 		return err
 	}
 
+	operatorClient, err := operatorv1client.NewForConfig(controller.KubeConfig)
+	if err != nil {
+		return err
+	}
+
 	gatherProtoKubeConfig, gatherKubeConfig, metricsGatherKubeConfig, alertsGatherKubeConfig := prepareGatherConfigs(
 		controller.ProtoKubeConfig, controller.KubeConfig, s.Impersonate,
 	)
@@ -114,7 +120,7 @@ func (s *Operator) Run(ctx context.Context, controller *controllercmd.Controller
 		gatherKubeConfig, gatherProtoKubeConfig, metricsGatherKubeConfig, alertsGatherKubeConfig, anonymizer,
 		configObserver, insightsClient,
 	)
-	periodicGather := periodic.New(configObserver, rec, gatherers, anonymizer)
+	periodicGather := periodic.New(configObserver, rec, gatherers, anonymizer, operatorClient.InsightsOperators())
 	statusReporter.AddSources(periodicGather.Sources()...)
 
 	// check we can read IO container status and we are not in crash loop

manifests/06-deployment.yaml manifests/05-deployment.yaml

@@ -4,28 +4,45 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 
+	v1 "github.com/openshift/api/operator/v1"
+	operatorv1client "github.com/openshift/client-go/operator/clientset/versioned/typed/operator/v1"
 	"github.com/openshift/insights-operator/pkg/anonymization"
 	"github.com/openshift/insights-operator/pkg/config/configobserver"
 	"github.com/openshift/insights-operator/pkg/controllerstatus"
 	"github.com/openshift/insights-operator/pkg/gather"
 	"github.com/openshift/insights-operator/pkg/gatherers"
 	"github.com/openshift/insights-operator/pkg/recorder"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	DataGatheredCondition = "DataGathered"
+	// NoDataGathered is a reason when there is no data gathered - e.g the resource is not in a cluster
+	NoDataGatheredReason = "NoData"
+	// GatherError is a reason when there is some error and no data gathered
+	GatherErrorReason = "Error"
+	// GatheredOK is a reason when data is gathered as expected
+	GatherOKReason = "GatheredOK"
+	// GatherWithError is a reason when data is gathered partially or with another error message
+	GatherWithErrorReason = "GatheredWithError"
 )
 
 // Controller periodically runs gatherers, records their results to the recorder
 // and flushes the recorder to create archives
 type Controller struct {
-	configurator configobserver.Configurator
-	recorder     recorder.FlushInterface
-	gatherers    []gatherers.Interface
-	statuses     map[string]controllerstatus.StatusController
-	anonymizer   *anonymization.Anonymizer
+	configurator        configobserver.Configurator
+	recorder            recorder.FlushInterface
+	gatherers           []gatherers.Interface
+	statuses            map[string]controllerstatus.StatusController
+	anonymizer          *anonymization.Anonymizer
+	insightsOperatorCLI operatorv1client.InsightsOperatorInterface
 }
 
 // New creates a new instance of Controller which periodically invokes the gatherers
@@ -35,6 +52,7 @@ func New(
 	rec recorder.FlushInterface,
 	listGatherers []gatherers.Interface,
 	anonymizer *anonymization.Anonymizer,
+	insightsOperatorCLI operatorv1client.InsightsOperatorInterface,
 ) *Controller {
 	statuses := make(map[string]controllerstatus.StatusController)
 
@@ -44,11 +62,12 @@ func New(
 	}
 
 	return &Controller{
-		configurator: configurator,
-		recorder:     rec,
-		gatherers:    listGatherers,
-		statuses:     statuses,
-		anonymizer:   anonymizer,
+		configurator:        configurator,
+		recorder:            rec,
+		gatherers:           listGatherers,
+		statuses:            statuses,
+		anonymizer:          anonymizer,
+		insightsOperatorCLI: insightsOperatorCLI,
 	}
 }
 
@@ -114,7 +133,7 @@ func (c *Controller) Gather() {
 	}
 
 	allFunctionReports := make(map[string]gather.GathererFunctionReport)
-
+	gatherTime := metav1.Now()
 	for _, gatherer := range gatherersToProcess {
 		func() {
 			name := gatherer.GetName()
@@ -142,8 +161,11 @@ func (c *Controller) Gather() {
 			})
 		}()
 	}
-
-	err := gather.RecordArchiveMetadata(mapToArray(allFunctionReports), c.recorder, c.anonymizer)
+	err := c.updateOperatorStatusCR(allFunctionReports, gatherTime)
+	if err != nil {
+		klog.Errorf("failed to update the Insights Operator CR status: %v", err)
+	}
+	err = gather.RecordArchiveMetadata(mapToArray(allFunctionReports), c.recorder, c.anonymizer)
 	if err != nil {
 		klog.Errorf("unable to record archive metadata because of error: %v", err)
 	}
@@ -176,6 +198,76 @@ func (c *Controller) periodicTrigger(stopCh <-chan struct{}) {
 	}
 }
 
+func (c *Controller) updateOperatorStatusCR(allFunctionReports map[string]gather.GathererFunctionReport, gatherTime metav1.Time) error {
+	insightsOperatorCR, err := c.insightsOperatorCLI.Get(context.Background(), "cluster", metav1.GetOptions{})
+	if err != nil {
+		return err
+	}
+
+	ioCopy := insightsOperatorCR.DeepCopy()
+	ioCopy.Status.GatherStatus.Gatherers = []v1.GathererStatus{}
+	ioCopy.Status.GatherStatus.LastGatherTime = gatherTime
+	ioCopy.Status.GatherStatus.LastGatherDuration = metav1.Duration{
+		Duration: time.Since(gatherTime.Time),
+	}
+
+	for k := range allFunctionReports {
+		fr := allFunctionReports[k]
+		// duration = 0 means the gatherer didn't run
+		if fr.Duration == 0 {
+			continue
+		}
+
+		gs := createGathererStatus(&fr)
+		ioCopy.Status.GatherStatus.Gatherers = append(ioCopy.Status.GatherStatus.Gatherers, gs)
+	}
+
+	_, err = c.insightsOperatorCLI.UpdateStatus(context.Background(), ioCopy, metav1.UpdateOptions{})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func createGathererStatus(gfr *gather.GathererFunctionReport) v1.GathererStatus {
+	gs := v1.GathererStatus{
+		Name: gfr.FuncName,
+		LastGatherDuration: metav1.Duration{
+			// v.Duration is in milliseconds and we need nanoseconds
+			Duration: time.Duration(gfr.Duration * 1000000),
+		},
+	}
+	con := metav1.Condition{
+		Type:               DataGatheredCondition,
+		LastTransitionTime: metav1.Now(),
+	}
+
+	if gfr.RecordsCount > 0 {
+		con.Status = metav1.ConditionTrue
+		con.Reason = GatherOKReason
+		con.Message = fmt.Sprintf("Created %d records in the archive.", gfr.RecordsCount)
+
+		if len(gfr.Errors) > 0 {
+			con.Reason = GatherWithErrorReason
+			con.Message = fmt.Sprintf("%s Error is: %s", con.Message, strings.Join(gfr.Errors, ","))
+		}
+
+		gs.Conditions = append(gs.Conditions, con)
+	} else {
+		con.Status = metav1.ConditionFalse
+		con.Reason = NoDataGatheredReason
+
+		if len(gfr.Errors) > 0 {
+			con.Reason = GatherErrorReason
+			con.Message = strings.Join(gfr.Errors, ",")
+		}
+
+		gs.Conditions = append(gs.Conditions, con)
+	}
+
+	return gs
+}
+
 func mapToArray(m map[string]gather.GathererFunctionReport) []gather.GathererFunctionReport {
 	a := make([]gather.GathererFunctionReport, 0, len(m))
 	for _, v := range m {

manifests/05-service.yaml manifests/06-service.yaml

@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	fakeOperatorCli "github.com/openshift/client-go/operator/clientset/versioned/fake"
 	"github.com/openshift/insights-operator/pkg/config"
 	"github.com/openshift/insights-operator/pkg/gather"
 	"github.com/openshift/insights-operator/pkg/gatherers"
@@ -172,6 +173,7 @@ func getMocksForPeriodicTest(listGatherers []gatherers.Interface, interval time.
 		Gather:   []string{gather.AllGatherersConst},
 	}}
 	mockRecorder := recorder.MockRecorder{}
-
-	return New(&mockConfigurator, &mockRecorder, listGatherers, nil), &mockRecorder
+	fakeInsightsOperatorCli := fakeOperatorCli.NewSimpleClientset().OperatorV1().InsightsOperators()
+	return New(&mockConfigurator, &mockRecorder, listGatherers, nil, fakeInsightsOperatorCli), &mockRecorder
 }
+