Implement periodic gathering as a job in tech preview (#787)

* Implement periodic gathering as a job in tech preview (#764)

* Run gathering as separate Pod in tech preview

* Move downloading of reports to operator part & propagate Insights Request ID

* Minor updates

* DataGather CR - very first  commit

* create a new DataGather CR and prune periodically

* read and apply gatherers config from the new CR

* Fix anonymizer

* do not duplicate gatherer status creation

* extract the job responsibility and fix contexts

* Copy Gatherer status & tests

* diskrecorder_test - narrow down the testing archive path

* Fix error reporting in periodic.go

* reorder manifest creation experiment

* status reporter must be always started for now

* rebase

* add resource requests to the new job

* rebase

* pass linting

* rebase

.golangci.yml

@@ -66,7 +66,7 @@ linters:
   enable:
     - bodyclose
     - deadcode
-    - depguard
+    #- depguard
     - dogsled
     - dupl
     - errcheck

cmd/insights-operator/main.go

@@ -60,6 +60,7 @@ func NewOperatorCommand() *cobra.Command {
 	cmd.AddCommand(start.NewOperator())
 	cmd.AddCommand(start.NewReceiver())
 	cmd.AddCommand(start.NewGather())
+	cmd.AddCommand(start.NewGatherAndUpload())
 
 	return cmd
 }

cmd/obfuscate-archive/main.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	configv1 "github.com/openshift/api/config/v1"
+	"github.com/openshift/api/insights/v1alpha1"
 	"github.com/openshift/insights-operator/pkg/anonymization"
 	"github.com/openshift/insights-operator/pkg/gather"
 	"github.com/openshift/insights-operator/pkg/record"
@@ -61,7 +62,7 @@ func obfuscateArchive(path string) (string, error) {
 		return "", err
 	}
 
-	anonymizer, err := anonymization.NewAnonymizer(clusterBaseDomain, networks, nil, nil, nil)
+	anonymizer, err := anonymization.NewAnonymizer(clusterBaseDomain, networks, nil, nil, v1alpha1.ObfuscateNetworking)
 	if err != nil {
 		return "", err
 	}

manifests/03-clusterrole.yaml

@@ -43,6 +43,19 @@ metadata:
     include.release.openshift.io/single-node-developer: "true"
     capability.openshift.io/name: Insights
 rules:
+  - apiGroups:
+      - "insights.openshift.io"
+    resources:
+      - datagathers
+      - datagathers/status
+    verbs:
+      - create
+      - get
+      - update
+      - patch
+      - list
+      - delete
+      - watch
   - apiGroups:
       - "operator.openshift.io"
     resources:
@@ -377,6 +390,22 @@ rules:
       - pods
     verbs:
       - get
+  - apiGroups:
+      - batch
+    resources:
+      - jobs
+    verbs:
+      - create
+      - get
+      - list 
+      - delete
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      - get
+
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding

pkg/anonymization/anonymizer.go

@@ -33,7 +33,7 @@ import (
 	"sync"
 
 	configv1 "github.com/openshift/api/config/v1"
-	"github.com/openshift/api/config/v1alpha1"
+	"github.com/openshift/api/insights/v1alpha1"
 	networkv1 "github.com/openshift/api/network/v1"
 	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
 	networkv1client "github.com/openshift/client-go/network/clientset/versioned/typed/network/v1"
@@ -88,7 +88,7 @@ type Anonymizer struct {
 	ipNetworkRegex     *regexp.Regexp
 	secretsClient      corev1client.SecretInterface
 	secretConfigurator configobserver.Configurator
-	apiConfigurator    configobserver.InsightsDataGatherObserver
+	dataPolicy         v1alpha1.DataPolicy
 	configClient       configv1client.ConfigV1Interface
 	networkClient      networkv1client.NetworkV1Interface
 	gatherKubeClient   kubernetes.Interface
@@ -104,7 +104,7 @@ func NewAnonymizer(clusterBaseDomain string,
 	networks []string,
 	secretsClient corev1client.SecretInterface,
 	secretConfigurator configobserver.Configurator,
-	apiConfigurator configobserver.InsightsDataGatherObserver) (*Anonymizer, error) {
+	dataPolicy v1alpha1.DataPolicy) (*Anonymizer, error) {
 	cidrs, err := k8snet.ParseCIDRs(networks)
 	if err != nil {
 		return nil, err
@@ -126,7 +126,7 @@ func NewAnonymizer(clusterBaseDomain string,
 		ipNetworkRegex:     regexp.MustCompile(Ipv4AddressOrNetworkRegex),
 		secretsClient:      secretsClient,
 		secretConfigurator: secretConfigurator,
-		apiConfigurator:    apiConfigurator,
+		dataPolicy:         dataPolicy,
 	}, nil
 }
 
@@ -138,14 +138,14 @@ func NewAnonymizerFromConfigClient(
 	configClient configv1client.ConfigV1Interface,
 	networkClient networkv1client.NetworkV1Interface,
 	secretConfigurator configobserver.Configurator,
-	apiConfigurator configobserver.InsightsDataGatherObserver,
+	dataPolicy v1alpha1.DataPolicy,
 ) (*Anonymizer, error) {
 	baseDomain, err := utils.GetClusterBaseDomain(ctx, configClient)
 	if err != nil {
 		return nil, err
 	}
 	secretsClient := kubeClient.CoreV1().Secrets(secretNamespace)
-	a, err := NewAnonymizer(baseDomain, []string{}, secretsClient, secretConfigurator, apiConfigurator)
+	a, err := NewAnonymizer(baseDomain, []string{}, secretsClient, secretConfigurator, dataPolicy)
 	if err != nil {
 		return nil, err
 	}
@@ -322,7 +322,7 @@ func NewAnonymizerFromConfig(
 	gatherProtoKubeConfig *rest.Config,
 	protoKubeConfig *rest.Config,
 	secretConfigurator configobserver.Configurator,
-	apiConfigurator configobserver.InsightsDataGatherObserver,
+	dataPolicy v1alpha1.DataPolicy,
 ) (*Anonymizer, error) {
 	kubeClient, err := kubernetes.NewForConfig(protoKubeConfig)
 	if err != nil {
@@ -344,7 +344,7 @@ func NewAnonymizerFromConfig(
 		return nil, err
 	}
 
-	return NewAnonymizerFromConfigClient(ctx, kubeClient, gatherKubeClient, configClient, networkClient, secretConfigurator, apiConfigurator)
+	return NewAnonymizerFromConfigClient(ctx, kubeClient, gatherKubeClient, configClient, networkClient, secretConfigurator, dataPolicy)
 }
 
 // AnonymizeMemoryRecord takes record.MemoryRecord, removes the sensitive data from it and returns the same object
@@ -484,8 +484,8 @@ func (anonymizer *Anonymizer) IsObfuscationEnabled() bool {
 	if anonymizer.secretConfigurator.Config().EnableGlobalObfuscation {
 		return true
 	}
-	if anonymizer.apiConfigurator != nil {
-		return *anonymizer.apiConfigurator.GatherDataPolicy() == v1alpha1.ObfuscateNetworking
+	if anonymizer.dataPolicy != "" {
+		return anonymizer.dataPolicy == v1alpha1.ObfuscateNetworking
 	}
 	return false
 }

pkg/anonymization/anonymizer_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	configv1 "github.com/openshift/api/config/v1"
-	"github.com/openshift/api/config/v1alpha1"
+	"github.com/openshift/api/insights/v1alpha1"
 	networkv1 "github.com/openshift/api/network/v1"
 	configfake "github.com/openshift/client-go/config/clientset/versioned/fake"
 	networkfake "github.com/openshift/client-go/network/clientset/versioned/fake"
@@ -123,11 +123,8 @@ func getAnonymizer(t *testing.T) *Anonymizer {
 	mockSecretConfigurator := config.NewMockSecretConfigurator(&config.Controller{
 		EnableGlobalObfuscation: true,
 	})
-	mockAPIConfigurator := config.NewMockAPIConfigurator(&v1alpha1.GatherConfig{
-		DataPolicy: v1alpha1.ObfuscateNetworking,
-	})
 	anonymizer, err := NewAnonymizer(clusterBaseDomain,
-		networks, kubefake.NewSimpleClientset().CoreV1().Secrets(secretNamespace), mockSecretConfigurator, mockAPIConfigurator)
+		networks, kubefake.NewSimpleClientset().CoreV1().Secrets(secretNamespace), mockSecretConfigurator, v1alpha1.ObfuscateNetworking)
 	assert.NoError(t, err)
 
 	return anonymizer
@@ -332,10 +329,7 @@ func TestAnonymizer_NewAnonymizerFromConfigClient(t *testing.T) {
 		networkClient,
 		config.NewMockSecretConfigurator(&config.Controller{
 			EnableGlobalObfuscation: true,
-		}),
-		config.NewMockAPIConfigurator(&v1alpha1.GatherConfig{
-			DataPolicy: v1alpha1.ObfuscateNetworking,
-		}),
+		}), v1alpha1.ObfuscateNetworking,
 	)
 	assert.NoError(t, err)
 	assert.NotNil(t, anonymizer)

pkg/authorizer/clusterauthorizer/clusterauthorizer_test.go

@@ -100,7 +100,7 @@ func Test_Proxy(tt *testing.T) {
 			co2 := &config.MockSecretConfigurator{Conf: &config.Controller{HTTPConfig: tc.HTTPConfig}}
 			a := Authorizer{proxyFromEnvironment: nonCachedProxyFromEnvironment(), configurator: co2}
 			p := a.NewSystemOrConfiguredProxy()
-			req := httptest.NewRequest("GET", tc.RequestURL, nil)
+			req := httptest.NewRequest("GET", tc.RequestURL, http.NoBody)
 			urlRec, err := p(req)
 
 			if err != nil {

pkg/cmd/start/start.go

@@ -25,7 +25,11 @@ import (
 	"github.com/openshift/insights-operator/pkg/controller"
 )
 
-const serviceCACertPath = "/var/run/configmaps/service-ca-bundle/service-ca.crt"
+const (
+	serviceCACertPath    = "/var/run/configmaps/service-ca-bundle/service-ca.crt"
+	pbContentType        = "application/vnd.kubernetes.protobuf"
+	pbAcceptContentTypes = "application/vnd.kubernetes.protobuf,application/json"
+)
 
 // NewOperator create the commad for running the Insights Operator.
 func NewOperator() *cobra.Command {
@@ -78,6 +82,29 @@ func NewGather() *cobra.Command {
 	return cmd
 }
 
+func NewGatherAndUpload() *cobra.Command {
+	operator := &controller.GatherJob{
+		Controller: config.Controller{
+			ConditionalGathererEndpoint: "https://console.redhat.com/api/gathering/gathering_rules",
+			StoragePath:                 "/var/lib/insights-operator",
+			Interval:                    2 * time.Hour,
+			Endpoint:                    "https://console.redhat.com/api/ingress/v1/upload",
+			ReportEndpoint:              "https://console.redhat.com/api/insights-results-aggregator/v2/cluster/%s/reports",
+			ReportPullingDelay:          60 * time.Second,
+			ReportMinRetryTime:          10 * time.Second,
+			ReportPullingTimeout:        30 * time.Minute,
+		},
+	}
+	cfg := controllercmd.NewControllerCommandConfig("openshift-insights-operator", version.Get(), nil)
+	cmd := &cobra.Command{
+		Use:   "gather-and-upload",
+		Short: "Runs the data gathering as job, uploads the data, waits for Insights analysis report and ends",
+		Run:   runGatherAndUpload(operator, cfg),
+	}
+	cmd.Flags().AddFlagSet(cfg.NewCommand().Flags())
+	return cmd
+}
+
 // Starts a single gather, main responsibility is loading in the necessary configs.
 func runGather(operator *controller.GatherJob, cfg *controllercmd.ControllerCommandConfig) func(cmd *cobra.Command, args []string) {
 	return func(cmd *cobra.Command, args []string) {
@@ -115,8 +142,8 @@ func runGather(operator *controller.GatherJob, cfg *controllercmd.ControllerComm
 			}
 		}
 		protoConfig := rest.CopyConfig(clientConfig)
-		protoConfig.AcceptContentTypes = "application/vnd.kubernetes.protobuf,application/json"
-		protoConfig.ContentType = "application/vnd.kubernetes.protobuf"
+		protoConfig.AcceptContentTypes = pbAcceptContentTypes
+		protoConfig.ContentType = pbContentType
 
 		ctx, cancel := context.WithTimeout(context.Background(), operator.Interval)
 
@@ -219,3 +246,52 @@ func runOperator(operator *controller.Operator, cfg *controllercmd.ControllerCom
 		}
 	}
 }
+
+// Starts a single gather, main responsibility is loading in the necessary configs.
+func runGatherAndUpload(operator *controller.GatherJob,
+	cfg *controllercmd.ControllerCommandConfig) func(cmd *cobra.Command, args []string) {
+	return func(cmd *cobra.Command, args []string) {
+		if configArg := cmd.Flags().Lookup("config").Value.String(); len(configArg) == 0 {
+			klog.Exit("error: --config is required")
+		}
+		unstructured, _, _, err := cfg.Config()
+		if err != nil {
+			klog.Exit(err)
+		}
+		cont, err := config.LoadConfig(operator.Controller, unstructured.Object, config.ToDisconnectedController)
+		if err != nil {
+			klog.Exit(err)
+		}
+		operator.Controller = cont
+
+		var clientConfig *rest.Config
+		if kubeConfigPath := cmd.Flags().Lookup("kubeconfig").Value.String(); len(kubeConfigPath) > 0 {
+			kubeConfigBytes, err := os.ReadFile(kubeConfigPath) //nolint: govet
+			if err != nil {
+				klog.Exit(err)
+			}
+			kubeConfig, err := clientcmd.NewClientConfigFromBytes(kubeConfigBytes)
+			if err != nil {
+				klog.Exit(err)
+			}
+			clientConfig, err = kubeConfig.ClientConfig()
+			if err != nil {
+				klog.Exit(err)
+			}
+		} else {
+			clientConfig, err = rest.InClusterConfig()
+			if err != nil {
+				klog.Exit(err)
+			}
+		}
+		protoConfig := rest.CopyConfig(clientConfig)
+		protoConfig.AcceptContentTypes = pbAcceptContentTypes
+		protoConfig.ContentType = pbContentType
+
+		err = operator.GatherAndUpload(clientConfig, protoConfig)
+		if err != nil {
+			klog.Exit(err)
+		}
+		os.Exit(0)
+	}
+}

pkg/config/configobserver/insighgtsdatagather_observer.go

@@ -56,7 +56,7 @@ func NewInsightsDataGatherObserver(kubeConfig *rest.Config,
 	return c, nil
 }
 
-func (i *insightsDataGatherController) sync(ctx context.Context, scx factory.SyncContext) error {
+func (i *insightsDataGatherController) sync(ctx context.Context, _ factory.SyncContext) error {
 	insightDataGatherConf, err := i.configV1Alpha1Cli.InsightsDataGathers().Get(ctx, "cluster", metav1.GetOptions{})
 	if err != nil {
 		return err