Cluster transfer OCM controller (#571)

* Implement the new cluster transfer controller

* Update the logic to get only accepted cluster transfers

* Minor fixes after code review

Author: tremes

config/pod.yaml

@@ -15,5 +15,7 @@ pull_report:
 ocm:
   scaEndpoint: https://api.openshift.com/api/accounts_mgmt/v1/certificates
   scaInterval: "8h"
+  clusterTransferEndpoint: https://api.openshift.com/api/accounts_mgmt/v1/cluster_transfers/
+  clusterTransferInterval: "24h"
 gather:
   - ALL

go.mod

@@ -4,6 +4,7 @@ go 1.16
 
 require (
 	github.com/blang/semver/v4 v4.0.0
+	github.com/evanphx/json-patch v4.12.0+incompatible
 	github.com/go-logr/logr v1.2.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
 	github.com/google/go-cmp v0.5.6 // indirect

manifests/03-clusterrole.yaml

@@ -310,6 +310,7 @@ rules:
       - support
     verbs:
       - get
+      - update
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding

pkg/cmd/start/start.go

@@ -34,8 +34,10 @@ func NewOperator() *cobra.Command {
 			ReportMinRetryTime:          10 * time.Second,
 			ReportPullingTimeout:        30 * time.Minute,
 			OCMConfig: config.OCMConfig{
-				SCAInterval: 8 * time.Hour,
-				SCAEndpoint: "https://api.openshift.com/api/accounts_mgmt/v1/certificates",
+				SCAInterval:             8 * time.Hour,
+				SCAEndpoint:             "https://api.openshift.com/api/accounts_mgmt/v1/certificates",
+				ClusterTransferEndpoint: "https://api.openshift.com/api/accounts_mgmt/v1/cluster_transfers",
+				ClusterTransferInterval: 24 * time.Hour,
 			},
 		},
 	}

pkg/config/config.go

@@ -26,9 +26,11 @@ type Serialized struct {
 	Gather                  []string `json:"gather"`
 	EnableGlobalObfuscation bool     `json:"enableGlobalObfuscation"`
 	OCM                     struct {
-		SCAEndpoint string `json:"scaEndpoint"`
-		SCAInterval string `json:"scaInterval"`
-		SCADisabled bool   `json:"scaDisabled"`
+		SCAEndpoint             string `json:"scaEndpoint"`
+		SCAInterval             string `json:"scaInterval"`
+		SCADisabled             bool   `json:"scaDisabled"`
+		ClusterTransferEndpoint string `json:"clusterTransferEndpoint"`
+		ClusterTransferInterval string `json:"clusterTransferInterval"`
 	}
 }
 
@@ -75,9 +77,11 @@ type HTTPConfig struct {
 
 // OCMConfig configures the interval and endpoint for retrieving the data from OCM API
 type OCMConfig struct {
-	SCAInterval time.Duration
-	SCAEndpoint string
-	SCADisabled bool
+	SCAInterval             time.Duration
+	SCAEndpoint             string
+	SCADisabled             bool
+	ClusterTransferEndpoint string
+	ClusterTransferInterval time.Duration
 }
 
 type Converter func(s *Serialized, cfg *Controller) (*Controller, error)
@@ -157,6 +161,13 @@ func (c *Controller) mergeOCM(cfg *Controller) {
 		c.OCMConfig.SCAInterval = cfg.OCMConfig.SCAInterval
 	}
 	c.OCMConfig.SCADisabled = cfg.OCMConfig.SCADisabled
+
+	if len(cfg.OCMConfig.ClusterTransferEndpoint) > 0 {
+		c.OCMConfig.ClusterTransferEndpoint = cfg.OCMConfig.ClusterTransferEndpoint
+	}
+	if cfg.OCMConfig.ClusterTransferInterval > 0 {
+		c.OCMConfig.ClusterTransferInterval = cfg.OCMConfig.ClusterTransferInterval
+	}
 }
 
 func (c *Controller) mergeHTTP(cfg *Controller) {
@@ -171,11 +182,10 @@ func (c *Controller) mergeInterval(cfg *Controller) {
 
 // ToController creates/updates a config Controller according to the Serialized config.
 // Makes sure that the config is correct.
-func ToController(s *Serialized, cfg *Controller) (*Controller, error) { // nolint: gocyclo
+func ToController(s *Serialized, cfg *Controller) (*Controller, error) { // nolint: gocyclo, funlen
 	if cfg == nil {
 		cfg = &Controller{}
 	}
-
 	cfg.Report = s.Report
 	cfg.StoragePath = s.StoragePath
 	cfg.Endpoint = s.Endpoint
@@ -252,6 +262,16 @@ func ToController(s *Serialized, cfg *Controller) (*Controller, error) { // noli
 		}
 		cfg.OCMConfig.SCAInterval = i
 	}
+	if len(s.OCM.SCAEndpoint) > 0 {
+		cfg.OCMConfig.ClusterTransferEndpoint = s.OCM.ClusterTransferEndpoint
+	}
+	if len(s.OCM.ClusterTransferInterval) > 0 {
+		i, err := time.ParseDuration(s.OCM.ClusterTransferInterval)
+		if err != nil {
+			return nil, fmt.Errorf("OCM Cluster transfer interval must be a valid duration: %v", err)
+		}
+		cfg.OCMConfig.ClusterTransferInterval = i
+	}
 	return cfg, nil
 }
 

pkg/config/configobserver/config.go

@@ -146,4 +146,18 @@ func (c *Config) loadOCM(data map[string][]byte) {
 	if scaDisabled, ok := data["scaPullDisabled"]; ok {
 		c.OCMConfig.SCADisabled = strings.EqualFold(string(scaDisabled), "true")
 	}
+
+	if clusterTransferEndpoint, ok := data["clusterTransferEndpoint"]; ok {
+		c.OCMConfig.ClusterTransferEndpoint = string(clusterTransferEndpoint)
+	}
+	if clusterTransferInterval, ok := data["clusterTransferInterval"]; ok {
+		if newInterval, err := time.ParseDuration(string(clusterTransferInterval)); err == nil {
+			c.OCMConfig.ClusterTransferInterval = newInterval
+		} else {
+			klog.Warningf(
+				"secret contains an invalid value (%s) for clusterTransferInterval. Using previous value",
+				clusterTransferInterval,
+			)
+		}
+	}
 }

pkg/controller/operator.go

@@ -27,6 +27,7 @@ import (
 	"github.com/openshift/insights-operator/pkg/insights/insightsclient"
 	"github.com/openshift/insights-operator/pkg/insights/insightsreport"
 	"github.com/openshift/insights-operator/pkg/insights/insightsuploader"
+	"github.com/openshift/insights-operator/pkg/ocm/clustertransfer"
 	"github.com/openshift/insights-operator/pkg/ocm/sca"
 	"github.com/openshift/insights-operator/pkg/recorder"
 	"github.com/openshift/insights-operator/pkg/recorder/diskrecorder"
@@ -149,6 +150,11 @@ func (s *Operator) Run(ctx context.Context, controller *controllercmd.Controller
 		statusReporter.AddSources(scaController)
 		go scaController.Run()
 	}
+
+	clusterTransferController := clustertransfer.New(ctx, kubeClient.CoreV1(), configObserver, insightsClient)
+	statusReporter.AddSources(clusterTransferController)
+	go clusterTransferController.Run()
+
 	klog.Warning("started")
 
 	<-ctx.Done()

pkg/controller/status/conditions.go

@@ -14,6 +14,9 @@ const (
 	InsightsDownloadDegraded configv1.ClusterStatusConditionType = "InsightsDownloadDegraded"
 	// SCANotAvailable is a condition type providing info about unsuccessful SCA pull attempt from the OCM API
 	SCANotAvailable configv1.ClusterStatusConditionType = "SCANotAvailable"
+	// ClusterTransferFailed is a condition type providing info about unsuccessful pull attempt of the ClusterTransfer from the OCM API
+	// or unsuccessful pull-secret update
+	ClusterTransferFailed configv1.ClusterStatusConditionType = "ClusterTransferFailed"
 )
 
 type conditionsMap map[configv1.ClusterStatusConditionType]configv1.ClusterOperatorStatusCondition

pkg/controller/status/controller.go

@@ -166,7 +166,7 @@ func (c *Controller) merge(clusterOperator *configv1.ClusterOperator) *configv1.
 }
 
 // calculate the current controller status based on its given sources
-func (c *Controller) currentControllerStatus() (allReady bool, lastTransition time.Time) {
+func (c *Controller) currentControllerStatus() (allReady bool, lastTransition time.Time) { //nolint: gocyclo
 	var errorReason string
 	var errs []string
 
@@ -210,6 +210,14 @@ func (c *Controller) currentControllerStatus() (allReady bool, lastTransition ti
 				degradingFailure = true
 			}
 			c.ctrlStatus.setStatus(SCAPullStatus, summary.Reason, summary.Message)
+		} else if summary.Operation.Name == controllerstatus.PullingClusterTransfer.Name {
+			// mark as degraded only in case of HTTP 500 and higher
+			if summary.Operation.HTTPStatusCode >= 500 {
+				klog.V(4).Infof("Failed to pull the cluster transfer object within the threshold %d with exponential backoff. Marking as degraded.",
+					OCMAPIFailureCountThreshold)
+				degradingFailure = true
+			}
+			c.ctrlStatus.setStatus(ClusterTransferStatus, summary.Reason, summary.Message)
 		}
 
 		if degradingFailure {
@@ -356,6 +364,13 @@ func updateControllerConditions(cs *conditions, ctrlStatus *controllerStatus,
 	} else {
 		cs.removeCondition(SCANotAvailable)
 	}
+
+	// handler when ClusterTransfer pull from the OCM fails
+	if ss := ctrlStatus.getStatus(ClusterTransferStatus); ss != nil {
+		cs.setCondition(ClusterTransferFailed, configv1.ConditionTrue, ss.reason, ss.message, metav1.Time{Time: lastTransition})
+	} else {
+		cs.removeCondition(ClusterTransferFailed)
+	}
 }
 
 // update the current controller state by it status

pkg/controller/status/status.go

@@ -1,11 +1,12 @@
 package status
 
 const (
-	DisabledStatus = "disabled"
-	UploadStatus   = "upload"
-	DownloadStatus = "download"
-	ErrorStatus    = "error"
-	SCAPullStatus  = "scaPullStatus"
+	DisabledStatus        = "disabled"
+	UploadStatus          = "upload"
+	DownloadStatus        = "download"
+	ErrorStatus           = "error"
+	SCAPullStatus         = "scaPullStatus"
+	ClusterTransferStatus = "clusterTransferStatus"
 )
 
 type controllerStatus struct {

pkg/controllerstatus/controllerstatus.go

@@ -12,10 +12,12 @@ type Interface interface {
 }
 
 type Operation struct {
-	Name           string
+	Name           OperationName
 	HTTPStatusCode int
 }
 
+type OperationName string
+
 var (
 	// DownloadingReport specific flag for Smart Proxy report downloading process.
 	DownloadingReport = Operation{Name: "DownloadingReport"}
@@ -25,6 +27,8 @@ var (
 	GatheringReport = Operation{Name: "GatheringReport"}
 	// PullingSCACerts is specific operation for pulling the SCA certs data from the OCM API
 	PullingSCACerts = Operation{Name: "PullingSCACerts"}
+	// PullingClusterTransfer is an operator for pulling ClusterTransfer object from the OCM API endpoint
+	PullingClusterTransfer = Operation{Name: "PullingClusterTransfer"}
 )
 
 // Summary represents the status summary of an Operation

pkg/insights/insightsclient/requests.go

@@ -5,11 +5,12 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"k8s.io/klog/v2"
 	"mime/multipart"
 	"net/http"
 	"strconv"
 
+	"k8s.io/klog/v2"
+
 	"github.com/openshift/insights-operator/pkg/authorizer"
 )
 
@@ -245,3 +246,47 @@ func (c *Client) RecvGatheringRules(ctx context.Context, endpoint string) ([]byt
 
 	return io.ReadAll(resp.Body)
 }
+
+// RecvClusterTransfer performs a request to the OCM cluster transfer API.
+// It is a HTTP GET request with the `search` query parameter limiting the result
+// only for the one cluster and only for the `accepted` cluster transfers.
+func (c *Client) RecvClusterTransfer(endpoint string) ([]byte, error) {
+	cv, err := c.getClusterVersion()
+	if err != nil {
+		return nil, err
+	}
+	if cv == nil {
+		return nil, ErrWaitingForVersion
+	}
+	token, err := c.authorizer.Token()
+	if err != nil {
+		return nil, err
+	}
+	req, err := http.NewRequest(http.MethodGet, endpoint, http.NoBody)
+	if err != nil {
+		return nil, err
+	}
+	q := req.URL.Query()
+	searchQuery := fmt.Sprintf("cluster_uuid is '%s' and status is 'accepted'", cv.Spec.ClusterID)
+	q.Add("search", searchQuery)
+	req.URL.RawQuery = q.Encode()
+	req.Header.Set("Content-Type", "application/json")
+	c.client.Transport = clientTransport(c.authorizer)
+	authHeader := fmt.Sprintf("AccessToken %s:%s", cv.Spec.ClusterID, token)
+	req.Header.Set("Authorization", authHeader)
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("unable to retrieve cluster transfer data from %s: %v", endpoint, err)
+	}
+
+	if resp.StatusCode > 399 || resp.StatusCode < 200 {
+		return nil, ocmErrorMessage(resp)
+	}
+	defer func() {
+		if err := resp.Body.Close(); err != nil {
+			klog.Warningf("Failed to close response body: %v", err)
+		}
+	}()
+	return io.ReadAll(resp.Body)
+}