improve target lifecycle hanlding

Signed-off-by: Frank Jogeleit <[email protected]>

pkg/listener/send_result_test.go

@@ -44,6 +44,10 @@ func (c client) Validate(rep v1alpha2.ReportInterface, result v1alpha2.PolicyRep
 	return c.validated
 }
 
+func (c *client) Reset(_ context.Context) error {
+	return nil
+}
+
 func (c *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {
 	c.cleanupCalled = true
 }

pkg/listener/sync_results.go

@@ -1,6 +1,7 @@
 package listener
 
 import (
+	"context"
 	"sync"
 
 	"github.com/kyverno/policy-reporter/pkg/crd/api/policyreport/v1alpha2"
@@ -12,6 +13,8 @@ import (
 const SendSyncResults = "send_sync_results_listener"
 
 func NewSendSyncResultsListener(targets *target.Collection) report.SyncResultsListener {
+	targets.Reset(context.Background())
+
 	return func(rep v1alpha2.ReportInterface) {
 		clients := targets.SyncClients()
 		if len(clients) == 0 {

pkg/target/client.go

@@ -41,6 +41,8 @@ type Client interface {
 	Type() ClientType
 	// CleanUp old results if supported by the target
 	CleanUp(context.Context, v1alpha2.ReportInterface)
+	// Reset the current state in the related target
+	Reset(context.Context) error
 }
 
 type ResultFilterFactory struct {
@@ -236,6 +238,14 @@ func (c *BaseClient) SkipExistingOnStartup() bool {
 	return c.skipExistingOnStartup
 }
 
+func (c *BaseClient) Reset(_ context.Context) error {
+	return nil
+}
+
+func (c *BaseClient) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
+
+func (c *BaseClient) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
+
 func NewBaseClient(options ClientOptions) BaseClient {
 	return BaseClient{options.Name, options.SkipExistingOnStartup, options.ResultFilter, options.ReportFilter}
 }

pkg/target/collection.go

@@ -1,9 +1,11 @@
 package target
 
 import (
+	"context"
 	"sync"
 
 	"github.com/kyverno/policy-reporter/pkg/helper"
+	"go.uber.org/zap"
 )
 
 type TargetType = string
@@ -56,6 +58,16 @@ func (c *Collection) Update(t *Target) {
 	c.mx.Unlock()
 }
 
+func (c *Collection) Reset(ctx context.Context) {
+	clients := c.SyncClients()
+
+	for _, c := range clients {
+		if err := c.Reset(ctx); err != nil {
+			zap.L().Error("failed to reset target", zap.String("type", c.Type()), zap.String("name", c.Name()))
+		}
+	}
+}
+
 func (c *Collection) Targets() []*Target {
 	return helper.ToList(c.targets)
 }

pkg/target/elasticsearch/elasticsearch.go

@@ -1,7 +1,6 @@
 package elasticsearch
 
 import (
-	"context"
 	"time"
 
 	"github.com/kyverno/policy-reporter/pkg/crd/api/policyreport/v1alpha2"
@@ -98,10 +97,6 @@ func (e *client) Send(result v1alpha2.PolicyReportResult) {
 	http.ProcessHTTPResponse(e.Name(), resp, err)
 }
 
-func (e *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (e *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (e *client) Type() target.ClientType {
 	return target.SingleSend
 }

pkg/target/factory/factory.go

@@ -672,8 +672,6 @@ func (f *TargetFactory) CreateSecurityHubTarget(config, parent *target.Config[ta
 		Synchronize:  config.Config.Synchronize,
 	})
 
-	hub.ResetFindings(context.Background())
-
 	return &target.Target{
 		ID:           uuid.NewString(),
 		Type:         target.SecurityHub,

pkg/target/gcs/gcs.go

@@ -2,7 +2,6 @@ package gcs
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"fmt"
 	"time"
@@ -63,10 +62,6 @@ func (c *client) Send(result v1alpha2.PolicyReportResult) {
 	zap.L().Info(c.Name() + ": PUSH OK")
 }
 
-func (c *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (c *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (c *client) Type() target.ClientType {
 	return target.SingleSend
 }

pkg/target/googlechat/googlechat.go

@@ -2,7 +2,6 @@ package googlechat
 
 import (
 	"bytes"
-	"context"
 	"text/template"
 	"time"
 
@@ -220,10 +219,6 @@ func (e *client) Send(result v1alpha2.PolicyReportResult) {
 	http.ProcessHTTPResponse(e.Name(), resp, err)
 }
 
-func (e *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (e *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (e *client) Type() target.ClientType {
 	return target.SingleSend
 }

pkg/target/kinesis/kinesis.go

@@ -2,7 +2,6 @@ package kinesis
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"fmt"
 	"time"
@@ -60,10 +59,6 @@ func (c *client) Send(result v1alpha2.PolicyReportResult) {
 	zap.L().Info("PUSH OK", zap.String("name", c.Name()))
 }
 
-func (c *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (c *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (c *client) Type() target.ClientType {
 	return target.SingleSend
 }

pkg/target/loki/loki.go

@@ -1,7 +1,6 @@
 package loki
 
 import (
-	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -137,8 +136,6 @@ func (l *client) Type() target.ClientType {
 	return target.BatchSend
 }
 
-func (l *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
 // NewClient creates a new loki.client to send Results to Loki
 func NewClient(options Options) target.Client {
 	return &client{

pkg/target/s3/s3.go

@@ -2,7 +2,6 @@ package s3
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"fmt"
 	"time"
@@ -62,10 +61,6 @@ func (c *client) Send(result v1alpha2.PolicyReportResult) {
 	zap.L().Info(c.Name() + ": PUSH OK")
 }
 
-func (c *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (c *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (c *client) Type() target.ClientType {
 	return target.SingleSend
 }

pkg/target/securityhub/securityhub.go

@@ -184,11 +184,12 @@ func (c *client) BatchSend(polr v1alpha2.ReportInterface, results []v1alpha2.Pol
 	zap.L().Info(c.Name()+": PUSH OK", zap.Int32("imported", *res.SuccessCount), zap.Int32("failed", *res.FailedCount), zap.String("report", polr.GetKey()))
 }
 
-func (c *client) ResetFindings(ctx context.Context) error {
+func (c *client) Reset(ctx context.Context) error {
 	if !c.synchronize {
 		return nil
 	}
-	defer zap.L().Info(c.Name() + ": START SYNC")
+
+	zap.L().Info(c.Name() + ": START SYNC")
 
 	list, err := c.getFindings(ctx)
 	if err != nil {

pkg/target/slack/slack.go

@@ -1,7 +1,6 @@
 package slack
 
 import (
-	"context"
 	"fmt"
 
 	"github.com/slack-go/slack"
@@ -313,8 +312,6 @@ func (s *client) Type() target.ClientType {
 	return target.BatchSend
 }
 
-func (s *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
 // NewClient creates a new slack.client to send Results to Slack
 func NewClient(options Options) target.Client {
 	return &client{

pkg/target/telegram/telegram.go

@@ -147,6 +147,10 @@ func (e *client) Send(result v1alpha2.PolicyReportResult) {
 
 func (e *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
 
+func (e *client) Reset(_ context.Context) error {
+	return nil
+}
+
 func (e *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
 
 func (e *client) Type() target.ClientType {

pkg/target/webhook/webhook.go

@@ -1,8 +1,6 @@
 package webhook
 
 import (
-	"context"
-
 	"github.com/kyverno/policy-reporter/pkg/crd/api/policyreport/v1alpha2"
 	"github.com/kyverno/policy-reporter/pkg/target"
 	"github.com/kyverno/policy-reporter/pkg/target/http"
@@ -53,10 +51,6 @@ func (e *client) Send(result v1alpha2.PolicyReportResult) {
 	http.ProcessHTTPResponse(e.Name(), resp, err)
 }
 
-func (e *client) CleanUp(_ context.Context, _ v1alpha2.ReportInterface) {}
-
-func (e *client) BatchSend(_ v1alpha2.ReportInterface, _ []v1alpha2.PolicyReportResult) {}
-
 func (e *client) Type() target.ClientType {
 	return target.SingleSend
 }