Bug 1954640: Support of gatherers with different periods

pkg/config/config.go

@@ -37,7 +37,16 @@ type Controller struct {
 	ReportMinRetryTime   time.Duration
 	ReportPullingTimeout time.Duration
 	Impersonate          string
-	Gather               []string
+	// list of gathering functions to enable
+	// if there's a string "ALL", we enable everything
+	// otherwise, each string should consist of 2 parts:
+	// gatherer name and function name split by slash
+	// Example: []{
+	//   "clusterconfig/container_images",
+	//   "clusterconfig/nodes",
+	//   "clusterconfig/authentication",
+	// }
+	Gather []string
 	// EnableGlobalObfuscation enables obfuscation of domain names and IP addresses
 	// To see the detailed info about how anonymization works, go to the docs of package anonymization.
 	EnableGlobalObfuscation bool

pkg/config/configobserver/configobserver.go

@@ -22,6 +22,11 @@ type ConfigReporter interface {
 	SetConfig(*config.Controller)
 }
 
+type Configurator interface {
+	Config() *config.Controller
+	ConfigChanged() (<-chan struct{}, func())
+}
+
 // Responsible for periodically checking and (if necessary) updating the local configs/tokens
 // according to the configs/tokens present on the cluster.
 type Controller struct {

pkg/config/mock_configurator.go

@@ -0,0 +1,13 @@
+package config
+
+// MockConfigurator returns the config from conf field
+type MockConfigurator struct {
+	Conf *Controller
+}
+
+func (mc *MockConfigurator) Config() *Controller {
+	return mc.Conf
+}
+func (mc *MockConfigurator) ConfigChanged() (<-chan struct{}, func()) {
+	return nil, nil
+}

pkg/controller/gather_job.go

@@ -16,7 +16,7 @@ import (
 	"github.com/openshift/insights-operator/pkg/anonymization"
 	"github.com/openshift/insights-operator/pkg/config"
 	"github.com/openshift/insights-operator/pkg/config/configobserver"
-	"github.com/openshift/insights-operator/pkg/gather/clusterconfig"
+	"github.com/openshift/insights-operator/pkg/gather"
 	"github.com/openshift/insights-operator/pkg/recorder"
 	"github.com/openshift/insights-operator/pkg/recorder/diskrecorder"
 )
@@ -69,8 +69,7 @@ func (d *GatherJob) Gather(ctx context.Context, kubeConfig, protoKubeConfig *res
 
 	var anonymizer *anonymization.Anonymizer
 	if anonymization.IsObfuscationEnabled(configObserver) {
-		var configClient *configv1client.ConfigV1Client
-		configClient, err = configv1client.NewForConfig(kubeConfig)
+		configClient, err := configv1client.NewForConfig(kubeConfig)
 		if err != nil {
 			return err
 		}
@@ -86,12 +85,18 @@ func (d *GatherJob) Gather(ctx context.Context, kubeConfig, protoKubeConfig *res
 	rec := recorder.New(recdriver, d.Interval, anonymizer)
 	defer rec.Flush()
 
-	// the gatherers check the state of the cluster and report any
-	// config to the recorder
-	clusterConfigGatherer := clusterconfig.New(gatherKubeConfig, gatherProtoKubeConfig, metricsGatherKubeConfig, anonymizer)
-	err = clusterConfigGatherer.Gather(ctx, configObserver.Config().Gather, rec)
-	if err != nil {
-		return err
+	gatherers := gather.CreateAllGatherers(
+		gatherKubeConfig, gatherProtoKubeConfig, metricsGatherKubeConfig, anonymizer,
+	)
+
+	var allFunctionReports []gather.GathererFunctionReport
+	for _, gatherer := range gatherers {
+		functionReports, err := gather.CollectAndRecordGatherer(ctx, gatherer, rec, configObserver)
+		if err != nil {
+			return err
+		}
+		allFunctionReports = append(allFunctionReports, functionReports...)
 	}
-	return nil
+
+	return gather.RecordArchiveMetadata(allFunctionReports, rec, anonymizer)
 }

pkg/controller/operator.go

@@ -26,7 +26,6 @@ import (
 	"github.com/openshift/insights-operator/pkg/controller/periodic"
 	"github.com/openshift/insights-operator/pkg/controller/status"
 	"github.com/openshift/insights-operator/pkg/gather"
-	"github.com/openshift/insights-operator/pkg/gather/clusterconfig"
 	"github.com/openshift/insights-operator/pkg/insights/insightsclient"
 	"github.com/openshift/insights-operator/pkg/insights/insightsreport"
 	"github.com/openshift/insights-operator/pkg/insights/insightsuploader"
@@ -120,14 +119,12 @@ func (s *Operator) Run(ctx context.Context, controller *controllercmd.Controller
 	rec := recorder.New(recdriver, s.Interval, anonymizer)
 	go rec.PeriodicallyPrune(ctx, statusReporter)
 
-	// the gatherers periodically check the state of the cluster and report any
-	// config to the recorder
-	clusterConfigGatherer := clusterconfig.New(
+	// the gatherers are periodically called to collect the data from the cluster
+	// and provide the results for the recorder
+	gatherers := gather.CreateAllGatherers(
 		gatherKubeConfig, gatherProtoKubeConfig, metricsGatherKubeConfig, anonymizer,
 	)
-	periodicGather := periodic.New(configObserver, rec, map[string]gather.Interface{
-		"clusterconfig": clusterConfigGatherer,
-	})
+	periodicGather := periodic.New(configObserver, rec, gatherers, anonymizer)
 	statusReporter.AddSources(periodicGather.Sources()...)
 
 	// check we can read IO container status and we are not in crash loop
@@ -168,9 +165,10 @@ func (s *Operator) Run(ctx context.Context, controller *controllercmd.Controller
 	reportGatherer := insightsreport.New(insightsClient, configObserver, uploader)
 	go reportGatherer.Run(ctx)
 
-	klog.Warning("stopped")
+	klog.Warning("started")
 
 	<-ctx.Done()
+
 	return nil
 }
 

pkg/controller/periodic/periodic.go

@@ -6,42 +6,51 @@ import (
 	"sort"
 	"time"
 
-	"k8s.io/klog/v2"
-
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
 
-	"github.com/openshift/insights-operator/pkg/config"
+	"github.com/openshift/insights-operator/pkg/anonymization"
+	"github.com/openshift/insights-operator/pkg/config/configobserver"
 	"github.com/openshift/insights-operator/pkg/controller/status"
 	"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"
 )
 
-type Configurator interface {
-	Config() *config.Controller
-	ConfigChanged() (<-chan struct{}, func())
-}
-
+// Controller periodically runs gatherers, records their results to the recorder
+// and flushes the recorder to create archives
 type Controller struct {
-	configurator Configurator
+	configurator configobserver.Configurator
 	recorder     recorder.FlushInterface
-	gatherers    map[string]gather.Interface
+	gatherers    []gatherers.Interface
 	statuses     map[string]*controllerstatus.Simple
+	anonymizer   *anonymization.Anonymizer
 }
 
-func New(configurator Configurator, rec recorder.FlushInterface, gatherers map[string]gather.Interface) *Controller {
+// New creates a new instance of Controller which periodically invokes the gatherers
+// and flushes the recorder to create archives.
+func New(
+	configurator configobserver.Configurator,
+	rec recorder.FlushInterface,
+	gatherers []gatherers.Interface,
+	anonymizer *anonymization.Anonymizer,
+) *Controller {
 	statuses := make(map[string]*controllerstatus.Simple)
-	for k := range gatherers {
-		statuses[k] = &controllerstatus.Simple{Name: fmt.Sprintf("periodic-%s", k)}
+
+	for _, gatherer := range gatherers {
+		gathererName := gatherer.GetName()
+		statuses[gathererName] = &controllerstatus.Simple{Name: fmt.Sprintf("periodic-%s", gathererName)}
 	}
-	c := &Controller{
+
+	return &Controller{
 		configurator: configurator,
 		recorder:     rec,
 		gatherers:    gatherers,
 		statuses:     statuses,
+		anonymizer:   anonymizer,
 	}
-	return c
 }
 
 func (c *Controller) Sources() []controllerstatus.Interface {
@@ -98,15 +107,46 @@ func (c *Controller) Gather() {
 		Steps:    threshold,
 		Cap:      interval,
 	}
-	for name := range c.gatherers {
+
+	// flush when all necessary gatherers were processed
+	defer func() {
+		if err := c.recorder.Flush(); err != nil {
+			klog.Errorf("Unable to flush the recorder: %v", err)
+		}
+	}()
+
+	var gatherersToProcess []gatherers.Interface
+
+	for _, gatherer := range c.gatherers {
+		if g, ok := gatherer.(gatherers.CustomPeriodGatherer); ok {
+			if g.ShouldBeProcessedNow() {
+				gatherersToProcess = append(gatherersToProcess, g)
+				g.UpdateLastProcessingTime()
+			}
+		} else {
+			gatherersToProcess = append(gatherersToProcess, gatherer)
+		}
+	}
+
+	var allFunctionReports []gather.GathererFunctionReport
+
+	for _, gatherer := range gatherersToProcess {
 		_ = wait.ExponentialBackoff(backoff, func() (bool, error) {
+			name := gatherer.GetName()
 			start := time.Now()
-			err := c.runGatherer(name)
+
+			ctx, cancel := context.WithTimeout(context.Background(), c.configurator.Config().Interval/2)
+			defer cancel()
+
+			klog.V(4).Infof("Running %s gatherer", gatherer.GetName())
+			functionReports, err := gather.CollectAndRecordGatherer(ctx, gatherer, c.recorder, c.configurator)
+			allFunctionReports = append(allFunctionReports, functionReports...)
 			if err == nil {
 				klog.V(3).Infof("Periodic gather %s completed in %s", name, time.Since(start).Truncate(time.Millisecond))
 				c.statuses[name].UpdateStatus(controllerstatus.Summary{Healthy: true})
 				return true, nil
 			}
+
 			utilruntime.HandleError(fmt.Errorf("%v failed after %s with: %v", name, time.Since(start).Truncate(time.Millisecond), err))
 			c.statuses[name].UpdateStatus(
 				controllerstatus.Summary{
@@ -117,25 +157,11 @@ func (c *Controller) Gather() {
 			return false, nil
 		})
 	}
-}
 
-// Does the prep for running a gatherer then calls gatherer.Gather. (getting the context, cleaning the recorder)
-func (c *Controller) runGatherer(name string) error {
-	gatherer, ok := c.gatherers[name]
-	if !ok {
-		klog.V(2).Infof("No such gatherer %s", name)
-		return nil
+	err := gather.RecordArchiveMetadata(allFunctionReports, c.recorder, c.anonymizer)
+	if err != nil {
+		klog.Errorf("unable to record archive metadata because of error: %v", err)
 	}
-	timeoutDuration := c.configurator.Config().Interval / 2
-	ctx, cancel := context.WithTimeout(context.Background(), timeoutDuration)
-	defer cancel()
-	defer func() {
-		if err := c.recorder.Flush(); err != nil {
-			klog.Errorf("Unable to flush recorder: %v", err)
-		}
-	}()
-	klog.V(4).Infof("Running %s", name)
-	return gatherer.Gather(ctx, c.configurator.Config().Gather, c.recorder)
 }
 
 // Periodically starts the gathering.

pkg/controller/periodic/periodic_test.go

@@ -0,0 +1,76 @@
+package periodic
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/openshift/insights-operator/pkg/config"
+	"github.com/openshift/insights-operator/pkg/gather"
+	"github.com/openshift/insights-operator/pkg/gatherers"
+	"github.com/openshift/insights-operator/pkg/recorder"
+)
+
+func Test_Controller_CustomPeriodGatherer(t *testing.T) {
+	c, mockRecorder := getMocksForPeriodicTest([]gatherers.Interface{
+		&gather.MockGatherer{CanFail: true},
+		&gather.MockCustomPeriodGatherer{Period: 999 * time.Hour},
+	})
+
+	c.Gather()
+	// 6 gatherers + metadata
+	assert.Len(t, mockRecorder.Records, 7)
+	mockRecorder.Reset()
+
+	c.Gather()
+	// 5 gatherers + metadata (one is low priority and we need to wait 999 hours to get it
+	assert.Len(t, mockRecorder.Records, 6)
+	mockRecorder.Reset()
+}
+
+func Test_Controller_CustomPeriodGathererNoPeriod(t *testing.T) {
+	mockGatherer := gather.MockCustomPeriodGathererNoPeriod{ShouldBeProcessed: true}
+	c, mockRecorder := getMocksForPeriodicTest([]gatherers.Interface{
+		&gather.MockGatherer{CanFail: true},
+		&mockGatherer,
+	})
+
+	c.Gather()
+	// 6 gatherers + metadata
+	assert.Len(t, mockRecorder.Records, 7)
+	assert.Equal(t, 1, mockGatherer.ShouldBeProcessedNowWasCalledNTimes)
+	assert.Equal(t, 1, mockGatherer.UpdateLastProcessingTimeWasCalledNTimes)
+	mockRecorder.Reset()
+
+	mockGatherer.ShouldBeProcessed = false
+
+	c.Gather()
+	// 5 gatherers + metadata (we've just disabled one gatherer)
+	assert.Len(t, mockRecorder.Records, 6)
+	assert.Equal(t, 2, mockGatherer.ShouldBeProcessedNowWasCalledNTimes)
+	// ShouldBeProcessedNow had returned false so we didn't call UpdateLastProcessingTime
+	assert.Equal(t, 1, mockGatherer.UpdateLastProcessingTimeWasCalledNTimes)
+	mockRecorder.Reset()
+
+	mockGatherer.ShouldBeProcessed = true
+
+	c.Gather()
+	assert.Len(t, mockRecorder.Records, 7)
+	assert.Equal(t, 3, mockGatherer.ShouldBeProcessedNowWasCalledNTimes)
+	assert.Equal(t, 2, mockGatherer.UpdateLastProcessingTimeWasCalledNTimes)
+	mockRecorder.Reset()
+}
+
+// TODO: cover more things
+
+func getMocksForPeriodicTest(gatherers []gatherers.Interface) (*Controller, *recorder.MockRecorder) {
+	mockConfigurator := config.MockConfigurator{Conf: &config.Controller{
+		Report:   true,
+		Interval: time.Hour,
+		Gather:   []string{gather.AllGatherersConst},
+	}}
+	mockRecorder := recorder.MockRecorder{}
+
+	return New(&mockConfigurator, &mockRecorder, gatherers, nil), &mockRecorder
+}