Development (#5)

* Implement loki.skipExistingOnStartup to prevent dulicated logs after deployment
* Implement loki.minimumPriority to configure which results should be send

charts/policy-reporter/templates/deployment.yaml

@@ -25,6 +25,15 @@ spec:
         - name: {{ .Chart.Name }}
           image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
           imagePullPolicy: {{ .Values.image.pullPolicy }}
+          args:
+            - run
+            - --loki={{ .Values.loki.host }}
+            {{- if .Values.loki.minimumPriority }}
+            - --loki-minimum-priority={{ .Values.loki.minimumPriority }}
+            {{- end }}
+            {{- if .Values.loki.skipExistingOnStartup }}
+            - --loki-skip-exising-on-startup
+            {{- end }}
           ports:
             - name: http
               containerPort: 2112
@@ -42,10 +51,7 @@ spec:
           volumeMounts:
           - name: config-volume
             mountPath: /app/config.yaml
-            subPath: config.yaml
-          env:
-          - name: LOKI_HOST
-            value: {{ .Values.loki.host | quote }}
+            subPath: config.yaml´
       volumes:
       - name: config-volume
         configMap:

charts/policy-reporter/values.yaml

@@ -1,5 +1,10 @@
 loki:
+  # loki host address
   host: http://loki.loki-stack.svc.cluster.local:3100
+  # minimum priority "" < info < warning < error
+  minimumPriority: ""
+  # Skip already existing PolicyReportResults on startup
+  skipExistingOnStartup: false
 
 metrics:
   serviceMonitor: false

cmd/root.go

@@ -4,7 +4,6 @@ import (
 	"flag"
 	"log"
 	"net/http"
-	"strings"
 
 	"github.com/fjogeleit/policy-reporter/pkg/config"
 	"github.com/fjogeleit/policy-reporter/pkg/report"
@@ -25,10 +24,9 @@ const (
 
 func NewCLI() *cobra.Command {
 	rootCmd := &cobra.Command{
-		Use:          "policyreporter",
-		SilenceUsage: true,
-		Short:        "Kyverno Policy API",
-		Long:         `Kyverno Policy API and Monitoring`,
+		Use:   "run",
+		Short: "Kyverno Policy API",
+		Long:  `Kyverno Policy API and Monitoring`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			c, err := LoadConfig(cmd)
 			if err != nil {
@@ -47,7 +45,7 @@ func NewCLI() *cobra.Command {
 			if loki != nil {
 				go client.WatchRuleValidation(func(r report.Result) {
 					go loki.Send(r)
-				})
+				}, c.Loki.SkipExisting)
 			}
 
 			policyMetrics, err := resolver.PolicyReportMetrics()
@@ -72,7 +70,10 @@ func NewCLI() *cobra.Command {
 	}
 
 	rootCmd.PersistentFlags().StringP("kubeconfig", "k", "", "absolute path to the kubeconfig file")
-	rootCmd.PersistentFlags().StringP("loki", "l", "", "loki host: http://loki:3100")
+
+	rootCmd.PersistentFlags().String("loki", "", "loki host: http://loki:3100")
+	rootCmd.PersistentFlags().String("loki-minimum-priority", "", "Minimum Priority to send Results to Loki (info < warning < error)")
+	rootCmd.PersistentFlags().Bool("loki-skip-exising-on-startup", false, "Skip Results created before PolicyReporter started. Prevent duplicated sending after new deployment")
 
 	flag.Parse()
 
@@ -81,7 +82,6 @@ func NewCLI() *cobra.Command {
 
 func LoadConfig(cmd *cobra.Command) (*config.Config, error) {
 	v := viper.New()
-	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
 
 	cfgFile := ""
 
@@ -106,6 +106,13 @@ func LoadConfig(cmd *cobra.Command) (*config.Config, error) {
 	if flag := cmd.Flags().Lookup("loki"); flag != nil {
 		v.BindPFlag("loki.host", flag)
 	}
+	if flag := cmd.Flags().Lookup("loki-minimum-priority"); flag != nil {
+		v.BindPFlag("loki.minimumPriority", flag)
+	}
+	if flag := cmd.Flags().Lookup("loki-skip-exising-on-startup"); flag != nil {
+		v.BindPFlag("loki.skipExistingOnStartup", flag)
+	}
+
 	if flag := cmd.Flags().Lookup("kubeconfig"); flag != nil {
 		v.BindPFlag("kubeconfig", flag)
 	}

config.example.yaml

@@ -1,7 +1,3 @@
-loki:
-  # loki host address
-  host: http://loki.loki-stack.svc.cluster.local:3100
-
 # Mapping from policy -> priority
 # Spported:
 #   error

pkg/config/config.go

@@ -1,11 +1,11 @@
 package config
 
-import "github.com/fjogeleit/policy-reporter/pkg/report"
-
 type Config struct {
 	Loki struct {
-		Host string `mapstructure:"host"`
+		Host            string `mapstructure:"host"`
+		SkipExisting    bool   `mapstructure:"skipExistingOnStartup"`
+		MinimumPriority string `mapstructure:"minimumPriority"`
 	} `mapstructure:"loki"`
-	Kubeconfig       string                     `mapstructure:"kubeconfig"`
-	PolicyPriorities map[string]report.Priority `mapstructure:"policy_priorities"`
+	Kubeconfig       string            `mapstructure:"kubeconfig"`
+	PolicyPriorities map[string]string `mapstructure:"policy_priorities"`
 }

pkg/config/resolver.go

@@ -1,6 +1,8 @@
 package config
 
 import (
+	"time"
+
 	"github.com/fjogeleit/policy-reporter/pkg/kubernetes"
 	"github.com/fjogeleit/policy-reporter/pkg/metrics"
 	"github.com/fjogeleit/policy-reporter/pkg/target"
@@ -23,7 +25,7 @@ func (r *Resolver) KubernetesClient() (kubernetes.Client, error) {
 		return kubeClient, nil
 	}
 
-	return kubernetes.NewDynamicClient(r.config.Kubeconfig, r.config.PolicyPriorities)
+	return kubernetes.NewDynamicClient(r.config.Kubeconfig, r.config.PolicyPriorities, time.Now())
 }
 
 func (r *Resolver) LokiClient() target.Client {
@@ -35,7 +37,10 @@ func (r *Resolver) LokiClient() target.Client {
 		return nil
 	}
 
-	return loki.NewClient(r.config.Loki.Host)
+	return loki.NewClient(
+		r.config.Loki.Host,
+		r.config.Loki.MinimumPriority,
+	)
 }
 
 func (r *Resolver) PolicyReportMetrics() (metrics.Metrics, error) {

pkg/kubernetes/client.go

@@ -2,8 +2,10 @@ package kubernetes
 
 import (
 	"context"
+	"errors"
 	"log"
 	"sync"
+	"time"
 
 	"github.com/fjogeleit/policy-reporter/pkg/report"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -27,15 +29,16 @@ type WatchPolicyResultCallback = func(report.Result)
 type Client interface {
 	FetchPolicyReports() []report.PolicyReport
 	WatchPolicyReports(WatchPolicyReportCallback)
-	WatchRuleValidation(WatchPolicyResultCallback)
-	WatchClusterPolicyReports(cb WatchClusterPolicyReportCallback)
+	WatchRuleValidation(WatchPolicyResultCallback, bool)
+	WatchClusterPolicyReports(WatchClusterPolicyReportCallback)
 }
 
 type DynamicClient struct {
 	client             dynamic.Interface
 	policyCache        map[string]report.PolicyReport
 	clusterPolicyCache map[string]report.ClusterPolicyReport
-	priorityMap        map[string]report.Priority
+	priorityMap        map[string]string
+	startUp            time.Time
 }
 
 func (c *DynamicClient) FetchPolicyReports() []report.PolicyReport {
@@ -82,14 +85,19 @@ func (c *DynamicClient) WatchPolicyReports(cb WatchPolicyReportCallback) {
 	}
 }
 
-func (c *DynamicClient) WatchRuleValidation(cb WatchPolicyResultCallback) {
+func (c *DynamicClient) WatchRuleValidation(cb WatchPolicyResultCallback, skipExisting bool) {
 	wg := sync.WaitGroup{}
 	wg.Add(2)
 
-	go func() {
-		c.WatchPolicyReports(func(s watch.EventType, pr report.PolicyReport) {
-			switch s {
+	go func(skipExisting bool) {
+		c.WatchPolicyReports(func(e watch.EventType, pr report.PolicyReport) {
+			switch e {
 			case watch.Added:
+				if skipExisting && pr.CreationTimestamp.Before(c.startUp) {
+					c.policyCache[pr.GetIdentifier()] = pr
+					break
+				}
+
 				for _, result := range pr.Results {
 					cb(result)
 				}
@@ -108,12 +116,17 @@ func (c *DynamicClient) WatchRuleValidation(cb WatchPolicyResultCallback) {
 		})
 
 		wg.Done()
-	}()
+	}(skipExisting)
 
-	go func() {
+	go func(skipExisting bool) {
 		c.WatchClusterPolicyReports(func(s watch.EventType, cpr report.ClusterPolicyReport) {
 			switch s {
 			case watch.Added:
+				if skipExisting && cpr.CreationTimestamp.Before(c.startUp) {
+					c.clusterPolicyCache[cpr.GetIdentifier()] = cpr
+					break
+				}
+
 				for _, result := range cpr.Results {
 					cb(result)
 				}
@@ -132,12 +145,12 @@ func (c *DynamicClient) WatchRuleValidation(cb WatchPolicyResultCallback) {
 		})
 
 		wg.Done()
-	}()
+	}(skipExisting)
 
 	wg.Wait()
 }
 
-func NewDynamicClient(kubeconfig string, prioties map[string]report.Priority) (Client, error) {
+func NewDynamicClient(kubeconfig string, prioties map[string]string, startUp time.Time) (Client, error) {
 	var config *rest.Config
 	var err error
 
@@ -160,6 +173,7 @@ func NewDynamicClient(kubeconfig string, prioties map[string]report.Priority) (C
 		policyCache:        make(map[string]report.PolicyReport),
 		clusterPolicyCache: make(map[string]report.ClusterPolicyReport),
 		priorityMap:        prioties,
+		startUp:            startUp,
 	}, nil
 }
 
@@ -208,6 +222,13 @@ func (c *DynamicClient) mapClusterPolicyReport(reportMap map[string]interface{})
 		Results: make(map[string]report.Result),
 	}
 
+	creationTimestamp, err := c.mapCreationTime(reportMap)
+	if err == nil {
+		r.CreationTimestamp = creationTimestamp
+	} else {
+		r.CreationTimestamp = time.Now()
+	}
+
 	if rs, ok := reportMap["results"].([]interface{}); ok {
 		for _, resultItem := range rs {
 			res := c.mapResult(resultItem.(map[string]interface{}))
@@ -218,6 +239,18 @@ func (c *DynamicClient) mapClusterPolicyReport(reportMap map[string]interface{})
 	return r
 }
 
+func (c *DynamicClient) mapCreationTime(result map[string]interface{}) (time.Time, error) {
+	if metadata, ok := result["metadata"].(map[string]interface{}); ok {
+		if created, ok2 := metadata["creationTimestamp"].(string); ok2 {
+			return time.Parse("2006-01-02T15:04:05Z", created)
+		}
+
+		return time.Time{}, errors.New("No creationTimestamp provided")
+	}
+
+	return time.Time{}, errors.New("No metadata provided")
+}
+
 func (c *DynamicClient) mapResult(result map[string]interface{}) report.Result {
 	var resources []report.Resource
 
@@ -240,17 +273,21 @@ func (c *DynamicClient) mapResult(result map[string]interface{}) report.Result {
 		}
 	}
 
+	status := result["status"].(report.Status)
+
 	r := report.Result{
 		Message:   result["message"].(string),
 		Policy:    result["policy"].(string),
-		Status:    result["status"].(report.Status),
+		Status:    status,
 		Scored:    result["scored"].(bool),
-		Priority:  report.Alert,
+		Priority:  report.PriorityFromStatus(status),
 		Resources: resources,
 	}
 
-	if priority, ok := c.priorityMap[r.Policy]; ok {
-		r.Priority = priority
+	if r.Status == report.Error || r.Status == report.Fail {
+		if priority, ok := c.priorityMap[r.Policy]; ok {
+			r.Priority = report.NewPriority(priority)
+		}
 	}
 
 	if rule, ok := result["rule"]; ok {

pkg/report/model.go

@@ -1,10 +1,12 @@
 package report
 
-import "fmt"
+import (
+	"fmt"
+	"time"
+)
 
 type Status = string
 type Severity = string
-type Priority = string
 
 const (
 	Fail  Status = "fail"
@@ -17,12 +19,68 @@ const (
 	Medium Severity = "medium"
 	Heigh  Severity = "heigh"
 
-	Alert       Priority = "error"
-	Warning     Priority = "warning"
-	Information Priority = "info"
-	Debug       Priority = "debug"
+	defaultString = ""
+	debugString   = "debug"
+	infoString    = "info"
+	warningString = "warning"
+	errorString   = "error"
 )
 
+const (
+	DefaultPriority = iota
+	DebugPriority
+	InfoPriority
+	WarningPriority
+	ErrorPriority
+)
+
+type Priority int
+
+func (p Priority) String() string {
+	switch p {
+	case DebugPriority:
+		return debugString
+	case InfoPriority:
+		return infoString
+	case WarningPriority:
+		return warningString
+	case ErrorPriority:
+		return errorString
+	default:
+		return defaultString
+	}
+}
+
+func PriorityFromStatus(p Status) Priority {
+	switch p {
+	case Fail:
+		return ErrorPriority
+	case Error:
+		return ErrorPriority
+	case Warn:
+		return WarningPriority
+	case Pass:
+		return InfoPriority
+	default:
+		return DefaultPriority
+	}
+}
+
+func NewPriority(p string) Priority {
+	switch p {
+	case debugString:
+		return DebugPriority
+	case infoString:
+		return InfoPriority
+	case warningString:
+		return WarningPriority
+	case errorString:
+		return ErrorPriority
+	default:
+		return DefaultPriority
+	}
+}
+
 type Resource struct {
 	APIVersion string
 	Kind       string
@@ -65,10 +123,11 @@ type Summary struct {
 }
 
 type PolicyReport struct {
-	Name      string
-	Namespace string
-	Results   map[string]Result
-	Summary   Summary
+	Name              string
+	Namespace         string
+	Results           map[string]Result
+	Summary           Summary
+	CreationTimestamp time.Time
 }
 
 func (pr PolicyReport) GetIdentifier() string {
@@ -90,9 +149,10 @@ func (pr PolicyReport) GetNewValidation(or PolicyReport) []Result {
 }
 
 type ClusterPolicyReport struct {
-	Name    string
-	Results map[string]Result
-	Summary Summary
+	Name              string
+	Results           map[string]Result
+	Summary           Summary
+	CreationTimestamp time.Time
 }
 
 func (cr ClusterPolicyReport) GetIdentifier() string {

pkg/target/loki/loki.go

@@ -28,7 +28,7 @@ type entry struct {
 }
 
 func newLokiPayload(result report.Result) payload {
-	le := entry{Ts: time.Now().Format(time.RFC3339), Line: "[" + mapPriority(result) + "] " + result.Message}
+	le := entry{Ts: time.Now().Format(time.RFC3339), Line: "[" + strings.ToUpper(result.Priority.String()) + "] " + result.Message}
 	ls := stream{Entries: []entry{le}}
 
 	res := report.Resource{}
@@ -40,7 +40,7 @@ func newLokiPayload(result report.Result) payload {
 	var labels = []string{
 		"status=\"" + result.Status + "\"",
 		"policy=\"" + result.Policy + "\"",
-		"priority=\"" + result.Priority + "\"",
+		"priority=\"" + result.Priority.String() + "\"",
 		"source=\"kyverno\"",
 	}
 
@@ -65,20 +65,17 @@ func newLokiPayload(result report.Result) payload {
 	return payload{Streams: []stream{ls}}
 }
 
-func mapPriority(r report.Result) string {
-	if r.Status == report.Error || r.Status == report.Fail {
-		return strings.ToUpper(r.Priority)
-	}
-
-	return strings.ToUpper(report.Information)
-}
-
 type Client struct {
-	host   string
-	client *http.Client
+	host            string
+	minimumPriority string
+	client          *http.Client
 }
 
 func (l *Client) Send(result report.Result) {
+	if result.Priority < report.NewPriority(l.minimumPriority) {
+		return
+	}
+
 	payload := newLokiPayload(result)
 	body := new(bytes.Buffer)
 
@@ -114,9 +111,10 @@ func (l *Client) Send(result report.Result) {
 	}
 }
 
-func NewClient(host string) target.Client {
+func NewClient(host, minimumPriority string) target.Client {
 	return &Client{
 		host + "/api/prom/push",
+		minimumPriority,
 		&http.Client{},
 	}
 }