Merge pull request #117 from martinkunc/dont-delay-first

Bug 1841057: Skip the initial upload delay

pkg/controller/status/status.go

@@ -43,10 +43,11 @@ type Controller struct {
 	statusCh     chan struct{}
 	configurator Configurator
 
-	lock     sync.Mutex
-	sources  []controllerstatus.Interface
-	reported Reported
-	start    time.Time
+	lock             sync.Mutex
+	sources          []controllerstatus.Interface
+	reported         Reported
+	start            time.Time
+	safeInitialStart bool
 }
 
 func NewController(client configv1client.ConfigV1Interface, configurator Configurator, namespace string) *Controller {
@@ -92,6 +93,18 @@ func (c *Controller) SetLastReportedTime(at time.Time) {
 	c.triggerStatusUpdate()
 }
 
+func (c *Controller) SafeInitialStart() bool {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	return c.safeInitialStart
+}
+
+func (c *Controller) SetSafeInitialStart(safe bool) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.safeInitialStart = safe
+}
+
 func (c *Controller) AddSources(sources ...controllerstatus.Interface) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
@@ -350,15 +363,25 @@ func (c *Controller) updateStatus(initial bool) error {
 		}
 		existing = nil
 	}
-	if initial && existing != nil {
-		var reported Reported
-		if len(existing.Status.Extension.Raw) > 0 {
-			if err := json.Unmarshal(existing.Status.Extension.Raw, &reported); err != nil {
-				klog.Errorf("The initial operator extension status is invalid: %v", err)
+	safeInitialStart := false
+	if initial {
+		if existing != nil {
+			var reported Reported
+			if len(existing.Status.Extension.Raw) > 0 {
+				if err := json.Unmarshal(existing.Status.Extension.Raw, &reported); err != nil {
+					klog.Errorf("The initial operator extension status is invalid: %v", err)
+				}
 			}
+			c.SetLastReportedTime(reported.LastReportTime.Time.UTC())
+			if c := findOperatorStatusCondition(existing.Status.Conditions, configv1.OperatorDegraded); c == nil ||
+				c != nil && c.Status == configv1.ConditionFalse {
+				safeInitialStart = true
+			}
+		} else {
+			safeInitialStart = true
 		}
-		c.SetLastReportedTime(reported.LastReportTime.Time.UTC())
 	}
+	c.SetSafeInitialStart(safeInitialStart)
 
 	updated := c.merge(existing)
 	if existing == nil {

pkg/controller/status/status_test.go

@@ -0,0 +1,89 @@
+package status
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/klog"
+
+	configv1 "github.com/openshift/api/config/v1"
+	configfake "github.com/openshift/client-go/config/clientset/versioned/fake"
+	"github.com/openshift/insights-operator/pkg/config"
+	"github.com/openshift/insights-operator/pkg/config/configobserver"
+	"github.com/openshift/insights-operator/pkg/utils"
+	kubeclientfake "k8s.io/client-go/kubernetes/fake"
+)
+
+func TestSaveInitialStart(t *testing.T) {
+
+	tests := []struct {
+		name                     string
+		clusterOperator          *configv1.ClusterOperator
+		expErr                   error
+		initialRun               bool
+		expectedSafeInitialStart bool
+	}{
+		{
+			name:                     "Non-initial run is has upload delayed",
+			initialRun:               false,
+			expectedSafeInitialStart: false,
+		},
+		{
+			name:                     "Initial run with not existing Insights operator is not delayed",
+			initialRun:               true,
+			clusterOperator:          nil,
+			expectedSafeInitialStart: true,
+		},
+		{
+			name:       "Initial run with existing Insights operator which is degraded is delayed",
+			initialRun: true,
+			clusterOperator: &configv1.ClusterOperator{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "insights",
+				},
+				Status: configv1.ClusterOperatorStatus{Conditions: []configv1.ClusterOperatorStatusCondition{
+					{Type: configv1.OperatorDegraded, Status: configv1.ConditionTrue},
+				}},
+			},
+			expectedSafeInitialStart: false,
+		},
+		{
+			name:       "Initial run with existing Insights operator which is not degraded not delayed",
+			initialRun: true,
+			clusterOperator: &configv1.ClusterOperator{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "insights",
+				},
+				Status: configv1.ClusterOperatorStatus{Conditions: []configv1.ClusterOperatorStatusCondition{
+					{Type: configv1.OperatorDegraded, Status: configv1.ConditionFalse},
+				}},
+			},
+			expectedSafeInitialStart: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			klog.SetOutput(utils.NewTestLog(t).Writer())
+			operators := []runtime.Object{}
+			if tt.clusterOperator != nil {
+				operators = append(operators, tt.clusterOperator)
+			}
+			kubeclientsetclient := kubeclientfake.NewSimpleClientset()
+
+			client := configfake.NewSimpleClientset(operators...)
+			ctrl := &Controller{name: "insights", client: client.ConfigV1(), configurator: configobserver.New(config.Controller{Report: true}, kubeclientsetclient)}
+
+			err := ctrl.updateStatus(tt.initialRun)
+			isSafe := ctrl.SafeInitialStart()
+			if err != tt.expErr {
+				t.Fatalf("updateStatus returned unexpected error: %s Expected %s", err, tt.expErr)
+			}
+			if isSafe != tt.expectedSafeInitialStart {
+				t.Fatalf("unexpected SafeInitialStart was: %t Expected %t", isSafe, tt.expectedSafeInitialStart)
+			}
+		})
+	}
+}

pkg/insights/insightsuploader/insightsuploader.go

@@ -34,6 +34,7 @@ type Summarizer interface {
 type StatusReporter interface {
 	LastReportedTime() time.Time
 	SetLastReportedTime(time.Time)
+	SafeInitialStart() bool
 }
 
 type Controller struct {
@@ -80,6 +81,9 @@ func (c *Controller) Run(ctx context.Context) {
 			initialDelay = wait.Jitter(now.Sub(next), 1.2)
 		}
 	}
+	if c.reporter.SafeInitialStart() {
+		initialDelay = 0
+	}
 	klog.V(2).Infof("Reporting status periodically to %s every %s, starting in %s", cfg.Endpoint, interval, initialDelay.Truncate(time.Second))
 
 	wait.Until(func() {