feat(rules/k8s): refactor common rules.k8s logic into `common/kuber…

…netes` package (grafana#6592) Signed-off-by: hainenber <[email protected]> Co-authored-by: William Dumont <[email protected]>
BarunKGP · Mar 19, 2024 · a7153e4 · a7153e4
1 parent 7b3664c
commit a7153e4
Show file tree

Hide file tree

Showing 18 changed files with 210 additions and 544 deletions.
diff --git a/...l/component/loki/rules/kubernetes/diff.go → internal/component/common/kubernetes/diff.go b/...l/component/loki/rules/kubernetes/diff.go → internal/component/common/kubernetes/diff.go
@@ -1,4 +1,4 @@
-package rules
+package kubernetes
 
 import (
 	"bytes"
@@ -7,27 +7,27 @@ import (
 	"gopkg.in/yaml.v3" // Used for prometheus rulefmt compatibility instead of gopkg.in/yaml.v2
 )
 
-type ruleGroupDiffKind string
+type RuleGroupDiffKind string
 
 const (
-	ruleGroupDiffKindAdd    ruleGroupDiffKind = "add"
-	ruleGroupDiffKindRemove ruleGroupDiffKind = "remove"
-	ruleGroupDiffKindUpdate ruleGroupDiffKind = "update"
+	RuleGroupDiffKindAdd    RuleGroupDiffKind = "add"
+	RuleGroupDiffKindRemove RuleGroupDiffKind = "remove"
+	RuleGroupDiffKindUpdate RuleGroupDiffKind = "update"
 )
 
-type ruleGroupDiff struct {
-	Kind    ruleGroupDiffKind
+type RuleGroupDiff struct {
+	Kind    RuleGroupDiffKind
 	Actual  rulefmt.RuleGroup
 	Desired rulefmt.RuleGroup
 }
 
-type ruleGroupsByNamespace map[string][]rulefmt.RuleGroup
-type ruleGroupDiffsByNamespace map[string][]ruleGroupDiff
+type RuleGroupsByNamespace map[string][]rulefmt.RuleGroup
+type RuleGroupDiffsByNamespace map[string][]RuleGroupDiff
 
-func diffRuleState(desired, actual ruleGroupsByNamespace) ruleGroupDiffsByNamespace {
+func DiffRuleState(desired, actual RuleGroupsByNamespace) RuleGroupDiffsByNamespace {
 	seenNamespaces := map[string]bool{}
 
-	diff := make(ruleGroupDiffsByNamespace)
+	diff := make(RuleGroupDiffsByNamespace)
 
 	for namespace, desiredRuleGroups := range desired {
 		seenNamespaces[namespace] = true
@@ -55,8 +55,8 @@ func diffRuleState(desired, actual ruleGroupsByNamespace) ruleGroupDiffsByNamesp
 	return diff
 }
 
-func diffRuleNamespaceState(desired []rulefmt.RuleGroup, actual []rulefmt.RuleGroup) []ruleGroupDiff {
-	var diff []ruleGroupDiff
+func diffRuleNamespaceState(desired []rulefmt.RuleGroup, actual []rulefmt.RuleGroup) []RuleGroupDiff {
+	var diff []RuleGroupDiff
 
 	seenGroups := map[string]bool{}
 
@@ -70,17 +70,17 @@ desiredGroups:
 					continue desiredGroups
 				}
 
-				diff = append(diff, ruleGroupDiff{
-					Kind:    ruleGroupDiffKindUpdate,
+				diff = append(diff, RuleGroupDiff{
+					Kind:    RuleGroupDiffKindUpdate,
 					Actual:  actualRuleGroup,
 					Desired: desiredRuleGroup,
 				})
 				continue desiredGroups
 			}
 		}
 
-		diff = append(diff, ruleGroupDiff{
-			Kind:    ruleGroupDiffKindAdd,
+		diff = append(diff, RuleGroupDiff{
+			Kind:    RuleGroupDiffKindAdd,
 			Desired: desiredRuleGroup,
 		})
 	}
@@ -90,8 +90,8 @@ desiredGroups:
 			continue
 		}
 
-		diff = append(diff, ruleGroupDiff{
-			Kind:   ruleGroupDiffKindRemove,
+		diff = append(diff, RuleGroupDiff{
+			Kind:   RuleGroupDiffKindRemove,
 			Actual: actualRuleGroup,
 		})
 	}

diff --git a/...ponent/loki/rules/kubernetes/diff_test.go → .../component/common/kubernetes/diff_test.go b/...ponent/loki/rules/kubernetes/diff_test.go → .../component/common/kubernetes/diff_test.go
@@ -1,4 +1,4 @@
-package rules
+package kubernetes
 
 import (
 	"fmt"
@@ -42,26 +42,26 @@ groups:
 		name     string
 		desired  map[string][]rulefmt.RuleGroup
 		actual   map[string][]rulefmt.RuleGroup
-		expected map[string][]ruleGroupDiff
+		expected map[string][]RuleGroupDiff
 	}
 
 	testCases := []testCase{
 		{
 			name:     "empty sets",
 			desired:  map[string][]rulefmt.RuleGroup{},
 			actual:   map[string][]rulefmt.RuleGroup{},
-			expected: map[string][]ruleGroupDiff{},
+			expected: map[string][]RuleGroupDiff{},
 		},
 		{
 			name: "add rule group",
 			desired: map[string][]rulefmt.RuleGroup{
 				managedNamespace: ruleGroupsA,
 			},
 			actual: map[string][]rulefmt.RuleGroup{},
-			expected: map[string][]ruleGroupDiff{
+			expected: map[string][]RuleGroupDiff{
 				managedNamespace: {
 					{
-						Kind:    ruleGroupDiffKindAdd,
+						Kind:    RuleGroupDiffKindAdd,
 						Desired: ruleGroupsA[0],
 					},
 				},
@@ -73,10 +73,10 @@ groups:
 			actual: map[string][]rulefmt.RuleGroup{
 				managedNamespace: ruleGroupsA,
 			},
-			expected: map[string][]ruleGroupDiff{
+			expected: map[string][]RuleGroupDiff{
 				managedNamespace: {
 					{
-						Kind:   ruleGroupDiffKindRemove,
+						Kind:   RuleGroupDiffKindRemove,
 						Actual: ruleGroupsA[0],
 					},
 				},
@@ -90,10 +90,10 @@ groups:
 			actual: map[string][]rulefmt.RuleGroup{
 				managedNamespace: ruleGroupsAModified,
 			},
-			expected: map[string][]ruleGroupDiff{
+			expected: map[string][]RuleGroupDiff{
 				managedNamespace: {
 					{
-						Kind:    ruleGroupDiffKindUpdate,
+						Kind:    RuleGroupDiffKindUpdate,
 						Desired: ruleGroupsA[0],
 						Actual:  ruleGroupsAModified[0],
 					},
@@ -108,28 +108,28 @@ groups:
 			actual: map[string][]rulefmt.RuleGroup{
 				managedNamespace: ruleGroupsA,
 			},
-			expected: map[string][]ruleGroupDiff{},
+			expected: map[string][]RuleGroupDiff{},
 		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			actual := diffRuleState(tc.desired, tc.actual)
+			actual := DiffRuleState(tc.desired, tc.actual)
 			requireEqualRuleDiffs(t, tc.expected, actual)
 		})
 	}
 }
 
-func requireEqualRuleDiffs(t *testing.T, expected, actual map[string][]ruleGroupDiff) {
+func requireEqualRuleDiffs(t *testing.T, expected, actual map[string][]RuleGroupDiff) {
 	require.Equal(t, len(expected), len(actual))
 
-	var summarizeDiff = func(diff ruleGroupDiff) string {
+	var summarizeDiff = func(diff RuleGroupDiff) string {
 		switch diff.Kind {
-		case ruleGroupDiffKindAdd:
+		case RuleGroupDiffKindAdd:
 			return fmt.Sprintf("add: %s", diff.Desired.Name)
-		case ruleGroupDiffKindRemove:
+		case RuleGroupDiffKindRemove:
 			return fmt.Sprintf("remove: %s", diff.Actual.Name)
-		case ruleGroupDiffKindUpdate:
+		case RuleGroupDiffKindUpdate:
 			return fmt.Sprintf("update: %s", diff.Desired.Name)
 		}
 		panic("unreachable")

diff --git a/internal/component/common/kubernetes/event.go b/internal/component/common/kubernetes/event.go
@@ -0,0 +1,61 @@
+package kubernetes
+
+import (
+	"github.com/go-kit/log"
+	"github.com/grafana/agent/internal/flow/logging/level"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+)
+
+// This type must be hashable, so it is kept simple. The indexer will maintain a
+// cache of current state, so this is mostly used for logging.
+type Event struct {
+	Typ       EventType
+	ObjectKey string
+}
+
+type EventType string
+
+const (
+	EventTypeResourceChanged EventType = "resource-changed"
+)
+
+type queuedEventHandler struct {
+	log   log.Logger
+	queue workqueue.RateLimitingInterface
+}
+
+func NewQueuedEventHandler(log log.Logger, queue workqueue.RateLimitingInterface) *queuedEventHandler {
+	return &queuedEventHandler{
+		log:   log,
+		queue: queue,
+	}
+}
+
+// OnAdd implements the cache.ResourceEventHandler interface.
+func (c *queuedEventHandler) OnAdd(obj interface{}, _ bool) {
+	c.publishEvent(obj)
+}
+
+// OnUpdate implements the cache.ResourceEventHandler interface.
+func (c *queuedEventHandler) OnUpdate(oldObj, newObj interface{}) {
+	c.publishEvent(newObj)
+}
+
+// OnDelete implements the cache.ResourceEventHandler interface.
+func (c *queuedEventHandler) OnDelete(obj interface{}) {
+	c.publishEvent(obj)
+}
+
+func (c *queuedEventHandler) publishEvent(obj interface{}) {
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		level.Error(c.log).Log("msg", "failed to get key for object", "err", err)
+		return
+	}
+
+	c.queue.AddRateLimited(Event{
+		Typ:       EventTypeResourceChanged,
+		ObjectKey: key,
+	})
+}
diff --git a/internal/component/common/kubernetes/rules.go b/internal/component/common/kubernetes/rules.go
@@ -0,0 +1,34 @@
+package kubernetes
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+type LabelSelector struct {
+	MatchLabels      map[string]string `river:"match_labels,attr,optional"`
+	MatchExpressions []MatchExpression `river:"match_expression,block,optional"`
+}
+
+type MatchExpression struct {
+	Key      string   `river:"key,attr"`
+	Operator string   `river:"operator,attr"`
+	Values   []string `river:"values,attr,optional"`
+}
+
+func ConvertSelectorToListOptions(selector LabelSelector) (labels.Selector, error) {
+	matchExpressions := []metav1.LabelSelectorRequirement{}
+
+	for _, me := range selector.MatchExpressions {
+		matchExpressions = append(matchExpressions, metav1.LabelSelectorRequirement{
+			Key:      me.Key,
+			Operator: metav1.LabelSelectorOperator(me.Operator),
+			Values:   me.Values,
+		})
+	}
+
+	return metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
+		MatchLabels:      selector.MatchLabels,
+		MatchExpressions: matchExpressions,
+	})
+}
diff --git a/internal/component/common/kubernetes/rules_test.go b/internal/component/common/kubernetes/rules_test.go
@@ -0,0 +1,13 @@
+package kubernetes
+
+import (
+	"testing"
+
+	"k8s.io/client-go/util/workqueue"
+)
+
+func TestEventTypeIsHashable(t *testing.T) {
+	// This test is here to ensure that the EventType type is hashable according to the workqueue implementation
+	queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+	queue.AddRateLimited(Event{})
+}