Merge pull request #185 from arangodb/bugfix/correlate-events

Use standard EventRecord to use event compression

main.go

@@ -39,11 +39,11 @@ import (
 	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/scheme"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/record"
 
 	"github.com/arangodb/kube-arangodb/pkg/client"
+	scheme "github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned/scheme"
 	"github.com/arangodb/kube-arangodb/pkg/logging"
 	"github.com/arangodb/kube-arangodb/pkg/operator"
 	"github.com/arangodb/kube-arangodb/pkg/server"

pkg/deployment/deployment.go

@@ -32,10 +32,9 @@ import (
 	"github.com/arangodb/arangosync/client"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
-	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/deployment/chaos"
@@ -60,6 +59,7 @@ type Dependencies struct {
 	Log           zerolog.Logger
 	KubeCli       kubernetes.Interface
 	DatabaseCRCli versioned.Interface
+	EventRecorder record.EventRecorder
 }
 
 // deploymentEventType strongly typed type of event
@@ -96,8 +96,6 @@ type Deployment struct {
 	stopCh  chan struct{}
 	stopped int32
 
-	eventsCli corev1.EventInterface
-
 	inspectTrigger            trigger.Trigger
 	updateDeploymentTrigger   trigger.Trigger
 	clientCache               *clientCache
@@ -121,7 +119,6 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeployment) (*De
 		deps:        deps,
 		eventCh:     make(chan *deploymentEvent, deploymentEventQueueSize),
 		stopCh:      make(chan struct{}),
-		eventsCli:   deps.KubeCli.Core().Events(apiObject.GetNamespace()),
 		clientCache: newClientCache(deps.KubeCli, apiObject),
 	}
 	d.status.last = *(apiObject.Status.DeepCopy())
@@ -337,11 +334,8 @@ func (d *Deployment) handleArangoDeploymentUpdatedEvent() error {
 
 // CreateEvent creates a given event.
 // On error, the error is logged.
-func (d *Deployment) CreateEvent(evt *v1.Event) {
-	_, err := d.eventsCli.Create(evt)
-	if err != nil {
-		d.deps.Log.Error().Err(err).Interface("event", *evt).Msg("Failed to record event")
-	}
+func (d *Deployment) CreateEvent(evt *k8sutil.Event) {
+	d.deps.EventRecorder.Event(evt.InvolvedObject, evt.Type, evt.Reason, evt.Message)
 }
 
 // Update the status of the API object from the internal status

pkg/deployment/members.go

@@ -28,7 +28,6 @@ import (
 
 	"github.com/dchest/uniuri"
 	"github.com/rs/zerolog"
-	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
@@ -42,7 +41,7 @@ func (d *Deployment) createInitialMembers(apiObject *api.ArangoDeployment) error
 	log.Debug().Msg("creating initial members...")
 
 	// Go over all groups and create members
-	var events []*v1.Event
+	var events []*k8sutil.Event
 	status, lastVersion := d.GetStatus()
 	if err := apiObject.ForeachServerGroup(func(group api.ServerGroup, spec api.ServerGroupSpec, members *api.MemberStatusList) error {
 		for len(*members) < spec.GetCount() {

pkg/deployment/reconcile/context.go

@@ -59,7 +59,7 @@ type Context interface {
 	GetSyncServerClient(ctx context.Context, group api.ServerGroup, id string) (client.API, error)
 	// CreateEvent creates a given event.
 	// On error, the error is logged.
-	CreateEvent(evt *v1.Event)
+	CreateEvent(evt *k8sutil.Event)
 	// CreateMember adds a new member to the given group.
 	// If ID is non-empty, it will be used, otherwise a new ID is created.
 	// Returns ID, error

pkg/deployment/resources/context.go

@@ -64,7 +64,7 @@ type Context interface {
 	GetNamespace() string
 	// CreateEvent creates a given event.
 	// On error, the error is logged.
-	CreateEvent(evt *v1.Event)
+	CreateEvent(evt *k8sutil.Event)
 	// GetOwnedPods returns a list of all pods owned by the deployment.
 	GetOwnedPods() ([]v1.Pod, error)
 	// GetOwnedPVCs returns a list of all PVCs owned by the deployment.

pkg/deployment/resources/pod_inspector.go

@@ -27,7 +27,6 @@ import (
 	"fmt"
 	"time"
 
-	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
@@ -47,7 +46,7 @@ const (
 // the member status of the deployment accordingly.
 func (r *Resources) InspectPods(ctx context.Context) error {
 	log := r.log
-	var events []*v1.Event
+	var events []*k8sutil.Event
 
 	pods, err := r.context.GetOwnedPods()
 	if err != nil {

pkg/operator/operator_deployment.go

@@ -212,6 +212,7 @@ func (o *Operator) makeDeploymentConfigAndDeps(apiObject *api.ArangoDeployment)
 			Logger(),
 		KubeCli:       o.Dependencies.KubeCli,
 		DatabaseCRCli: o.Dependencies.CRCli,
+		EventRecorder: o.Dependencies.EventRecorder,
 	}
 	return cfg, deps
 }

pkg/operator/operator_deployment_relication.go

@@ -208,8 +208,9 @@ func (o *Operator) makeDeploymentReplicationConfigAndDeps(apiObject *api.ArangoD
 		Log: o.Dependencies.LogService.MustGetLogger("deployment-replication").With().
 			Str("deployment-replication", apiObject.GetName()).
 			Logger(),
-		KubeCli: o.Dependencies.KubeCli,
-		CRCli:   o.Dependencies.CRCli,
+		KubeCli:       o.Dependencies.KubeCli,
+		CRCli:         o.Dependencies.CRCli,
+		EventRecorder: o.Dependencies.EventRecorder,
 	}
 	return cfg, deps
 }

pkg/operator/operator_local_storage.go

@@ -210,8 +210,9 @@ func (o *Operator) makeLocalStorageConfigAndDeps(apiObject *api.ArangoLocalStora
 		Log: o.Dependencies.LogService.MustGetLogger("storage").With().
 			Str("localStorage", apiObject.GetName()).
 			Logger(),
-		KubeCli:      o.Dependencies.KubeCli,
-		StorageCRCli: o.Dependencies.CRCli,
+		KubeCli:       o.Dependencies.KubeCli,
+		StorageCRCli:  o.Dependencies.CRCli,
+		EventRecorder: o.Dependencies.EventRecorder,
 	}
 	return cfg, deps
 }

pkg/replication/deployment_replication.go

@@ -29,10 +29,9 @@ import (
 	"time"
 
 	"github.com/rs/zerolog"
-	"k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
 
 	"github.com/arangodb/arangosync/client"
 	api "github.com/arangodb/kube-arangodb/pkg/apis/replication/v1alpha"
@@ -49,9 +48,10 @@ type Config struct {
 
 // Dependencies holds dependent services for a DeploymentReplication
 type Dependencies struct {
-	Log     zerolog.Logger
-	KubeCli kubernetes.Interface
-	CRCli   versioned.Interface
+	Log           zerolog.Logger
+	KubeCli       kubernetes.Interface
+	CRCli         versioned.Interface
+	EventRecorder record.EventRecorder
 }
 
 // deploymentReplicationEvent strongly typed type of event
@@ -84,8 +84,6 @@ type DeploymentReplication struct {
 	stopCh  chan struct{}
 	stopped int32
 
-	eventsCli corev1.EventInterface
-
 	inspectTrigger         trigger.Trigger
 	recentInspectionErrors int
 	clientCache            client.ClientCache
@@ -103,7 +101,6 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoDeploymentReplic
 		deps:      deps,
 		eventCh:   make(chan *deploymentReplicationEvent, deploymentReplicationEventQueueSize),
 		stopCh:    make(chan struct{}),
-		eventsCli: deps.KubeCli.Core().Events(apiObject.GetNamespace()),
 	}
 
 	go dr.run()
@@ -241,10 +238,8 @@ func (dr *DeploymentReplication) handleArangoDeploymentReplicationUpdatedEvent(e
 
 // createEvent creates a given event.
 // On error, the error is logged.
-func (dr *DeploymentReplication) createEvent(evt *v1.Event) {
-	if _, err := dr.eventsCli.Create(evt); err != nil {
-		dr.deps.Log.Error().Err(err).Interface("event", *evt).Msg("Failed to record event")
-	}
+func (dr *DeploymentReplication) createEvent(evt *k8sutil.Event) {
+	dr.deps.EventRecorder.Event(evt.InvolvedObject, evt.Type, evt.Reason, evt.Message)
 }
 
 // Update the status of the API object from the internal status

pkg/storage/local_storage.go

@@ -35,6 +35,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/storage/v1alpha"
 	"github.com/arangodb/kube-arangodb/pkg/generated/clientset/versioned"
@@ -52,9 +53,10 @@ type Config struct {
 
 // Dependencies holds dependent services for a LocalStorage
 type Dependencies struct {
-	Log          zerolog.Logger
-	KubeCli      kubernetes.Interface
-	StorageCRCli versioned.Interface
+	Log           zerolog.Logger
+	KubeCli       kubernetes.Interface
+	StorageCRCli  versioned.Interface
+	EventRecorder record.EventRecorder
 }
 
 // localStorageEvent strongly typed type of event
@@ -112,7 +114,6 @@ func New(config Config, deps Dependencies, apiObject *api.ArangoLocalStorage) (*
 		deps:      deps,
 		eventCh:   make(chan *localStorageEvent, localStorageEventQueueSize),
 		stopCh:    make(chan struct{}),
-		eventsCli: deps.KubeCli.Core().Events(apiObject.GetNamespace()),
 	}
 
 	ls.pvCleaner = newPVCleaner(deps.Log, deps.KubeCli, ls.GetClientByNodeName)
@@ -324,11 +325,8 @@ func (ls *LocalStorage) handleArangoLocalStorageUpdatedEvent(event *localStorage
 
 // createEvent creates a given event.
 // On error, the error is logged.
-func (ls *LocalStorage) createEvent(evt *v1.Event) {
-	_, err := ls.eventsCli.Create(evt)
-	if err != nil {
-		ls.deps.Log.Error().Err(err).Interface("event", *evt).Msg("Failed to record event")
-	}
+func (ls *LocalStorage) createEvent(evt *k8sutil.Event) {
+	ls.deps.EventRecorder.Event(evt.InvolvedObject, evt.Type, evt.Reason, evt.Message)
 }
 
 // Update the status of the API object from the internal status