Policy report crd update (#24)

* Add Support for Properties and Timestamp
* Prevent resending violations after Kyverno cleanup

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 1.2.0
+
+* Support for (Cluster)PolicyReport CRD Properties in Target Output
+* Support for (Cluster)PolicyReport CRD Timestamp in Target Output
+* Fix resend violations after Kyverno Cleanup with ResultHashes
+
 ## 1.1.0
 
 * Added PolicyReport Category to Metrics

charts/policy-reporter/Chart.yaml

@@ -5,8 +5,8 @@ description: |
   It creates Prometheus Metrics and can send rule validation events to different targets like Loki, Elasticsearch, Slack or Discord
 
 type: application
-version: 1.1.0
-appVersion: 1.1.0
+version: 1.2.0
+appVersion: 1.2.0
 
 dependencies:
   - name: monitoring

charts/policy-reporter/values.yaml

@@ -1,7 +1,7 @@
 image:
   repository: fjogeleit/policy-reporter
   pullPolicy: IfNotPresent
-  tag: 1.1.0
+  tag: 1.2.0
 
 imagePullSecrets: []
 

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/magiconair/properties v1.8.4 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.1 // indirect
 	github.com/mitchellh/mapstructure v1.4.1 // indirect
 	github.com/pelletier/go-toml v1.8.1 // indirect
 	github.com/prometheus/client_golang v1.9.0

go.sum

@@ -292,6 +292,8 @@ github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrk
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
 github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS42BGNg=
+github.com/mitchellh/hashstructure/v2 v2.0.1 h1:L60q1+q7cXE4JeEJJKMnh2brFIe3rZxCihYAB61ypAY=
+github.com/mitchellh/hashstructure/v2 v2.0.1/go.mod h1:MG3aRVU/N29oo/V/IhBX8GR/zz4kQkprJgF2EVszyDE=
 github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY=
 github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=

pkg/kubernetes/cluster_policy_report_client.go

@@ -20,6 +20,7 @@ type clusterPolicyReportClient struct {
 	startUp         time.Time
 	skipExisting    bool
 	started         bool
+	modifyHash      map[string]uint64
 }
 
 func (c *clusterPolicyReportClient) RegisterCallback(cb report.ClusterPolicyReportCallback) {
@@ -125,6 +126,12 @@ func (c *clusterPolicyReportClient) RegisterPolicyResultWatcher(skipExisting boo
 	c.RegisterCallback(func(s watch.EventType, cpr report.ClusterPolicyReport, opr report.ClusterPolicyReport) {
 		switch s {
 		case watch.Added:
+			if len(cpr.Results) == 0 {
+				break
+			}
+
+			c.modifyHash[cpr.GetIdentifier()] = cpr.ResultHash()
+
 			preExisted := cpr.CreationTimestamp.Before(c.startUp)
 
 			if c.skipExisting && preExisted {
@@ -145,6 +152,19 @@ func (c *clusterPolicyReportClient) RegisterPolicyResultWatcher(skipExisting boo
 
 			wg.Wait()
 		case watch.Modified:
+			if len(cpr.Results) == 0 {
+				break
+			}
+
+			newHash := cpr.ResultHash()
+			if hash, ok := c.modifyHash[cpr.GetIdentifier()]; ok {
+				if newHash == hash {
+					break
+				}
+			}
+
+			c.modifyHash[cpr.GetIdentifier()] = newHash
+
 			diff := cpr.GetNewResults(opr)
 
 			wg := sync.WaitGroup{}
@@ -160,16 +180,21 @@ func (c *clusterPolicyReportClient) RegisterPolicyResultWatcher(skipExisting boo
 			}
 
 			wg.Wait()
+		case watch.Deleted:
+			if _, ok := c.modifyHash[cpr.GetIdentifier()]; ok {
+				delete(c.modifyHash, cpr.GetIdentifier())
+			}
 		}
 	})
 }
 
 // NewPolicyReportClient creates a new PolicyReportClient based on the kubernetes go-client
 func NewClusterPolicyReportClient(client PolicyReportAdapter, store *report.ClusterPolicyReportStore, mapper Mapper, startUp time.Time) report.ClusterPolicyClient {
 	return &clusterPolicyReportClient{
-		policyAPI: client,
-		store:     store,
-		mapper:    mapper,
-		startUp:   startUp,
+		policyAPI:  client,
+		store:      store,
+		mapper:     mapper,
+		startUp:    startUp,
+		modifyHash: make(map[string]uint64),
 	}
 }

pkg/kubernetes/mapper.go

@@ -140,12 +140,13 @@ func (m *mapper) mapResult(result map[string]interface{}) report.Result {
 	status := result["status"].(report.Status)
 
 	r := report.Result{
-		Message:   result["message"].(string),
-		Policy:    result["policy"].(string),
-		Status:    status,
-		Scored:    result["scored"].(bool),
-		Priority:  report.PriorityFromStatus(status),
-		Resources: resources,
+		Message:    result["message"].(string),
+		Policy:     result["policy"].(string),
+		Status:     status,
+		Scored:     result["scored"].(bool),
+		Priority:   report.PriorityFromStatus(status),
+		Resources:  resources,
+		Properties: make(map[string]string, 0),
 	}
 
 	if severity, ok := result["severity"]; ok {
@@ -164,6 +165,21 @@ func (m *mapper) mapResult(result map[string]interface{}) report.Result {
 		r.Category = category.(string)
 	}
 
+	if created, ok := result["timestamp"]; ok {
+		time, err := time.Parse("2006-01-02T15:04:05Z", created.(string))
+		if err == nil {
+			r.Timestamp = time
+		}
+	}
+
+	if props, ok := result["properties"]; ok {
+		if properties, ok := props.(map[string]interface{}); ok {
+			for property, value := range properties {
+				r.Properties[property] = value.(string)
+			}
+		}
+	}
+
 	return r
 }
 

pkg/kubernetes/mapper_test.go

@@ -28,13 +28,14 @@ var policyMap = map[string]interface{}{
 	},
 	"results": []interface{}{
 		map[string]interface{}{
-			"message":  "message",
-			"status":   "fail",
-			"scored":   true,
-			"policy":   "required-label",
-			"rule":     "app-label-required",
-			"category": "test",
-			"severity": "high",
+			"message":   "message",
+			"status":    "fail",
+			"scored":    true,
+			"policy":    "required-label",
+			"rule":      "app-label-required",
+			"timestamp": "2021-02-23T15:10:00Z",
+			"category":  "test",
+			"severity":  "high",
 			"resources": []interface{}{
 				map[string]interface{}{
 					"apiVersion": "v1",
@@ -44,6 +45,9 @@ var policyMap = map[string]interface{}{
 					"uid":        "dfd57c50-f30c-4729-b63f-b1954d8988d1",
 				},
 			},
+			"properties": map[string]interface{}{
+				"version": "1.2.0",
+			},
 		},
 		map[string]interface{}{
 			"message":   "message 2",
@@ -163,6 +167,12 @@ func Test_MapPolicyReport(t *testing.T) {
 	if result1.Severity != report.High {
 		t.Errorf("Expected Severity '%s' (acutal %s)", report.High, result1.Severity)
 	}
+	if result1.Timestamp.Format("2006-01-02T15:04:05Z") != "2021-02-23T15:10:00Z" {
+		t.Errorf("Expected Timestamp '2021-02-23T15:10:00Z' (acutal %s)", result1.Timestamp.Format("2006-01-02T15:04:05Z"))
+	}
+	if result1.Properties["version"] != "1.2.0" {
+		t.Errorf("Expected Property '1.2.0' (acutal %s)", result1.Properties["version"])
+	}
 
 	resource := result1.Resources[0]
 	if resource.APIVersion != "v1" {

pkg/kubernetes/policy_report_client.go

@@ -20,6 +20,7 @@ type policyReportClient struct {
 	startUp         time.Time
 	skipExisting    bool
 	started         bool
+	modifyHash      map[string]uint64
 }
 
 func (c *policyReportClient) RegisterCallback(cb report.PolicyReportCallback) {
@@ -126,34 +127,75 @@ func (c *policyReportClient) RegisterPolicyResultWatcher(skipExisting bool) {
 		func(e watch.EventType, pr report.PolicyReport, or report.PolicyReport) {
 			switch e {
 			case watch.Added:
+				if len(pr.Results) == 0 {
+					break
+				}
+
+				c.modifyHash[pr.GetIdentifier()] = pr.ResultHash()
+
 				preExisted := pr.CreationTimestamp.Before(c.startUp)
 
 				if c.skipExisting && preExisted {
 					break
 				}
 
-				for _, result := range pr.Results {
+				wg := sync.WaitGroup{}
+				wg.Add(len(pr.Results) * len(c.resultCallbacks))
+
+				for _, r := range pr.Results {
 					for _, cb := range c.resultCallbacks {
-						cb(result, preExisted)
+						go func(callback report.PolicyResultCallback, result report.Result) {
+							callback(result, preExisted)
+							wg.Done()
+						}(cb, r)
 					}
 				}
+
+				wg.Wait()
 			case watch.Modified:
+				if len(pr.Results) == 0 {
+					break
+				}
+
+				newHash := pr.ResultHash()
+				if hash, ok := c.modifyHash[pr.GetIdentifier()]; ok {
+					if newHash == hash {
+						break
+					}
+				}
+
+				c.modifyHash[pr.GetIdentifier()] = newHash
+
 				diff := pr.GetNewResults(or)
-				for _, result := range diff {
+
+				wg := sync.WaitGroup{}
+				wg.Add(len(diff) * len(c.resultCallbacks))
+
+				for _, r := range diff {
 					for _, cb := range c.resultCallbacks {
-						cb(result, false)
+						go func(callback report.PolicyResultCallback, result report.Result) {
+							callback(result, false)
+							wg.Done()
+						}(cb, r)
 					}
 				}
+
+				wg.Wait()
+			case watch.Deleted:
+				if _, ok := c.modifyHash[pr.GetIdentifier()]; ok {
+					delete(c.modifyHash, pr.GetIdentifier())
+				}
 			}
 		})
 }
 
 // NewPolicyReportClient creates a new PolicyReportClient based on the kubernetes go-client
 func NewPolicyReportClient(client PolicyReportAdapter, store *report.PolicyReportStore, mapper Mapper, startUp time.Time) report.PolicyClient {
 	return &policyReportClient{
-		policyAPI: client,
-		store:     store,
-		mapper:    mapper,
-		startUp:   startUp,
+		policyAPI:  client,
+		store:      store,
+		mapper:     mapper,
+		startUp:    startUp,
+		modifyHash: make(map[string]uint64),
 	}
 }