Skip to content

Commit

Permalink
🌱 Persistent the agent event state to the configmap (stolostron#899)
Browse files Browse the repository at this point in the history
* Add time filter

Signed-off-by: myan <[email protected]>

* add more tests

Signed-off-by: myan <[email protected]>

* format

Signed-off-by: Meng Yan <[email protected]>

* refactor test

Signed-off-by: myan <[email protected]>

* fix sonar

Signed-off-by: myan <[email protected]>

* reply review

Signed-off-by: myan <[email protected]>

* use creattionTimestamp

Signed-off-by: myan <[email protected]>

---------

Signed-off-by: myan <[email protected]>
Signed-off-by: Meng Yan <[email protected]>
  • Loading branch information
yanmxa authored May 9, 2024
1 parent a98da42 commit 81b5143
Show file tree
Hide file tree
Showing 11 changed files with 551 additions and 166 deletions.
6 changes: 6 additions & 0 deletions agent/pkg/status/controller/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/apps"
agentstatusconfig "github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/config"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/event"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/filter"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/hubcluster"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/managedclusters"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/placement"
Expand Down Expand Up @@ -77,5 +78,10 @@ func AddControllers(ctx context.Context, mgr ctrl.Manager, agentConfig *config.A
if err := apps.LaunchSubscriptionReportSyncer(ctx, mgr, agentConfig, producer); err != nil {
return fmt.Errorf("failed to launch subscription report syncer: %w", err)
}

// lunch a time filter, it must be called after filter.RegisterTimeFilter(key)
if err := filter.LaunchTimeFilter(ctx, mgr.GetClient(), agentConfig.PodNameSpace); err != nil {
return fmt.Errorf("failed to launch time filter: %w", err)
}
return nil
}
135 changes: 3 additions & 132 deletions agent/pkg/status/controller/event/event_integration_test.go
Original file line number Diff line number Diff line change
@@ -1,139 +1,10 @@
package event

import (
"encoding/json"
"fmt"

. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
corev1 "k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
clusterv1 "open-cluster-management.io/api/cluster/v1"
policyv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/stolostron/multicluster-global-hub/pkg/bundle/event"
"github.com/stolostron/multicluster-global-hub/pkg/constants"
"github.com/stolostron/multicluster-global-hub/pkg/enum"
)

var _ = Describe("test the policy emitter", Ordered, func() {
It("should pass the root policy event", func() {
By("Creating a root policy")
rootPolicy := &policyv1.Policy{
ObjectMeta: metav1.ObjectMeta{
Name: "policy1",
Namespace: "default",
Finalizers: []string{constants.GlobalHubCleanupFinalizer},
},
Spec: policyv1.PolicySpec{
Disabled: true,
PolicyTemplates: []*policyv1.PolicyTemplate{},
},
Status: policyv1.PolicyStatus{
ComplianceState: policyv1.Compliant,
},
}
Expect(kubeClient.Create(ctx, rootPolicy)).NotTo(HaveOccurred())

evt := &corev1.Event{
ObjectMeta: metav1.ObjectMeta{
Name: "policy1.123r543243242",
Namespace: "default",
},
InvolvedObject: v1.ObjectReference{
Kind: string(policyv1.Kind),
Namespace: "default",
Name: "policy1",
},
Reason: "PolicyPropagation",
Message: "Policy default/policy1 was propagated to cluster1",
Source: corev1.EventSource{
Component: "policy-propagator",
},
}
Expect(kubeClient.Create(ctx, evt)).NotTo(HaveOccurred())

receivedEvent := <-consumer.EventChan()
fmt.Sprintln(receivedEvent)
Expect(string(enum.LocalRootPolicyEventType)).To(Equal(receivedEvent.Type()))

rootPolicyEvents := []event.RootPolicyEvent{}
err := json.Unmarshal(receivedEvent.Data(), &rootPolicyEvents)
Expect(err).Should(Succeed())
Expect(rootPolicyEvents[0].EventName).To(Equal(evt.Name))
})

It("should pass the replicated policy event", func() {
By("Create namespace and cluster for the replicated policy")
err := kubeClient.Create(ctx, &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: "cluster1",
},
}, &client.CreateOptions{})
Expect(err).Should(Succeed())

By("Create the cluster")
cluster := &clusterv1.ManagedCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "cluster1",
},
}
Expect(kubeClient.Create(ctx, cluster, &client.CreateOptions{})).Should(Succeed())
cluster.Status = clusterv1.ManagedClusterStatus{
ClusterClaims: []clusterv1.ManagedClusterClaim{
{
Name: "id.k8s.io",
Value: "3f406177-34b2-4852-88dd-ff2809680336",
},
},
}
Expect(kubeClient.Status().Update(ctx, cluster)).Should(Succeed())

By("Create the replicated policy")
replicatedPolicy := &policyv1.Policy{
ObjectMeta: metav1.ObjectMeta{
Name: "default.policy1",
Namespace: "cluster1",
Labels: map[string]string{
constants.PolicyEventRootPolicyNameLabelKey: fmt.Sprintf("%s.%s", "default", "policy1"),
constants.PolicyEventClusterNameLabelKey: "cluster1",
},
},
Spec: policyv1.PolicySpec{
Disabled: false,
PolicyTemplates: []*policyv1.PolicyTemplate{},
},
}
Expect(kubeClient.Create(ctx, replicatedPolicy)).ToNot(HaveOccurred())

By("Create the replicated policy event")
evt := &corev1.Event{
ObjectMeta: metav1.ObjectMeta{
Name: "default.policy1.17af98f19c06811e",
Namespace: "cluster1",
},
InvolvedObject: v1.ObjectReference{
Kind: string(policyv1.Kind),
Namespace: "cluster1",
Name: "default.policy1",
},
Reason: "PolicyStatusSync",
Message: "Policy default.policy1 status was updated in cluster",
Source: corev1.EventSource{
Component: "policy-status-sync",
},
}
Expect(kubeClient.Create(ctx, evt)).NotTo(HaveOccurred())

receivedEvent := <-consumer.EventChan()
fmt.Sprintln(receivedEvent)
Expect(string(enum.LocalReplicatedPolicyEventType)).To(Equal(receivedEvent.Type()))

replicatedPolicyEvents := []event.ReplicatedPolicyEvent{}
err = json.Unmarshal(receivedEvent.Data(), &replicatedPolicyEvents)
Expect(err).Should(Succeed())
Expect(replicatedPolicyEvents[0].EventName).To(Equal(evt.Name))
})
var _ = Describe("Events emitters", Ordered, func() {
Context("with root policy events", Ordered, localRootPolicyEventTestSpecs)
Context("with replicated policy events", Ordered, localReplicatedPolicyEventTestSpecs)
})
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,16 @@ package event
import (
"context"
"fmt"
"strings"

cloudevents "github.com/cloudevents/sdk-go/v2"
"github.com/go-logr/logr"
lru "github.com/hashicorp/golang-lru"
corev1 "k8s.io/api/core/v1"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/config"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/filter"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/generic"
"github.com/stolostron/multicluster-global-hub/agent/pkg/status/controller/policies"
"github.com/stolostron/multicluster-global-hub/pkg/bundle/event"
Expand Down Expand Up @@ -60,29 +61,30 @@ var _ generic.ObjectEmitter = &localReplicatedPolicyEmitter{}

type localReplicatedPolicyEmitter struct {
ctx context.Context
name string
log logr.Logger
eventType string
runtimeClient client.Client
currentVersion *version.Version
lastSentVersion version.Version
payload event.ReplicatedPolicyEventBundle
cache *lru.Cache
topic string
}

func NewLocalReplicatedPolicyEmitter(ctx context.Context, runtimeClient client.Client,
topic string,
) generic.ObjectEmitter {
cache, _ := lru.New(20)
name := strings.Replace(string(enum.LocalReplicatedPolicyEventType), enum.EventTypePrefix, "", -1)
filter.RegisterTimeFilter(name)
return &localReplicatedPolicyEmitter{
ctx: ctx,
log: ctrl.Log.WithName("policy-event-syncer/replicated-policy"),
name: name,
log: ctrl.Log.WithName(name),
eventType: string(enum.LocalReplicatedPolicyEventType),
topic: topic,
runtimeClient: runtimeClient,
currentVersion: version.NewVersion(),
lastSentVersion: *version.NewVersion(),
cache: cache,
payload: make([]event.ReplicatedPolicyEvent, 0),
}
}
Expand All @@ -96,7 +98,7 @@ func (h *localReplicatedPolicyEmitter) ShouldUpdate(obj client.Object) bool {
if config.GetEnableLocalPolicy() != config.EnableLocalPolicyTrue {
return false
}
policy, ok := policyEventPredicate(h.ctx, obj, h.runtimeClient, h.log)
policy, ok := policyEventPredicate(h.ctx, h.name, obj, h.runtimeClient, h.log)

return ok && !utils.HasAnnotation(policy, constants.OriginOwnerReferenceAnnotation) &&
utils.HasItemKey(policy.GetLabels(), constants.PolicyEventRootPolicyNameLabelKey)
Expand All @@ -115,11 +117,6 @@ func (h *localReplicatedPolicyEmitter) Update(obj client.Object) bool {
if !ok {
return false
}
// if exist, then return
evtKey := getEventKey(evt)
if h.cache.Contains(evtKey) {
return false
}

// get policy
policy, err := getInvolvePolicy(h.ctx, h.runtimeClient, evt)
Expand Down Expand Up @@ -151,7 +148,6 @@ func (h *localReplicatedPolicyEmitter) Update(obj client.Object) bool {
}
// cache to events and update version
h.payload = append(h.payload, replicatedPolicyEvent)
h.cache.Add(evtKey, nil)
return true
}

Expand All @@ -173,6 +169,10 @@ func (h *localReplicatedPolicyEmitter) ToCloudEvent() (*cloudevents.Event, error
}

func (h *localReplicatedPolicyEmitter) PostSend() {
// update the time filter: with latest event
for _, evt := range h.payload {
filter.CacheTime(h.name, evt.CreatedAt.Time)
}
// update version and clean the cache
h.payload = make([]event.ReplicatedPolicyEvent, 0)
h.currentVersion.Next()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
package event

import (
"encoding/json"
"fmt"
"time"

. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
corev1 "k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
clusterv1 "open-cluster-management.io/api/cluster/v1"
policyv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/stolostron/multicluster-global-hub/pkg/bundle/event"
"github.com/stolostron/multicluster-global-hub/pkg/constants"
"github.com/stolostron/multicluster-global-hub/pkg/enum"
)

func localReplicatedPolicyEventTestSpecs() {
It("should pass the replicated policy event", func() {
By("Create namespace and cluster for the replicated policy")
err := kubeClient.Create(ctx, &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: "cluster1",
},
}, &client.CreateOptions{})
Expect(err).Should(Succeed())

By("Create the cluster")
cluster := &clusterv1.ManagedCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "cluster1",
},
}
Expect(kubeClient.Create(ctx, cluster, &client.CreateOptions{})).Should(Succeed())
cluster.Status = clusterv1.ManagedClusterStatus{
ClusterClaims: []clusterv1.ManagedClusterClaim{
{
Name: "id.k8s.io",
Value: "3f406177-34b2-4852-88dd-ff2809680336",
},
},
}
Expect(kubeClient.Status().Update(ctx, cluster)).Should(Succeed())

By("Create the replicated policy")
replicatedPolicy := &policyv1.Policy{
ObjectMeta: metav1.ObjectMeta{
Name: "default.policy1",
Namespace: "cluster1",
Labels: map[string]string{
constants.PolicyEventRootPolicyNameLabelKey: fmt.Sprintf("%s.%s", "default", "policy1"),
constants.PolicyEventClusterNameLabelKey: "cluster1",
},
},
Spec: policyv1.PolicySpec{
Disabled: false,
PolicyTemplates: []*policyv1.PolicyTemplate{},
},
}
Expect(kubeClient.Create(ctx, replicatedPolicy)).ToNot(HaveOccurred())

By("Create the replicated policy event")
evt := &corev1.Event{
ObjectMeta: metav1.ObjectMeta{
Name: "default.policy1.17af98f19c06811e",
Namespace: "cluster1",
},
InvolvedObject: v1.ObjectReference{
Kind: string(policyv1.Kind),
Namespace: "cluster1",
Name: "default.policy1",
},
Reason: "PolicyStatusSync",
Message: "Policy default.policy1 status was updated in cluster",
Source: corev1.EventSource{
Component: "policy-status-sync",
},
LastTimestamp: metav1.Time{Time: time.Now()},
}
Expect(kubeClient.Create(ctx, evt)).NotTo(HaveOccurred())

receivedEvent := <-consumer.EventChan()
fmt.Println(">>>>>>>>>>>>>>>>>>> replicated policy event", receivedEvent)
Expect(string(enum.LocalReplicatedPolicyEventType)).To(Equal(receivedEvent.Type()))

replicatedPolicyEvents := []event.ReplicatedPolicyEvent{}
err = json.Unmarshal(receivedEvent.Data(), &replicatedPolicyEvents)
Expect(err).Should(Succeed())
Expect(replicatedPolicyEvents[0].EventName).To(Equal(evt.Name))
})
}
Loading

0 comments on commit 81b5143

Please sign in to comment.