test: add events tests (#27)

Makefile

@@ -10,7 +10,7 @@ GOLINT = golangci-lint run
 # Source repository variables.
 ROOT_DIR := $(shell git rev-parse --show-toplevel)
 BIN_DIR = $(ROOT_DIR)/bin
-TEST_PKGS = $(shell go list ./...)
+TEST_PKGS = $(shell go list ./... | grep -v 'github.com/azure/peerd/api\|github.com/azure/peerd/pkg/mocks') # Exclude generated and mock code.
 TESTS_BIN_DIR = $(BIN_DIR)/tests
 COVERAGE_DIR=$(BIN_DIR)/coverage
 SCRIPTS_DIR=$(ROOT_DIR)/build/ci/scripts

cmd/proxy/main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/azure/peerd/internal/routing"
 	"github.com/azure/peerd/internal/state"
 	"github.com/azure/peerd/pkg/containerd"
+	"github.com/azure/peerd/pkg/k8s"
 	"github.com/rs/zerolog"
 	"github.com/spf13/afero"
 	"golang.org/x/sync/errgroup"
@@ -78,7 +79,12 @@ func serverCommand(ctx context.Context, args *ServerCmd) (err error) {
 		return err
 	}
 
-	ctx, err = events.WithContext(ctx)
+	clientset, err := k8s.NewKubernetesInterface(p2pcontext.KubeConfigPath)
+	if err != nil {
+		return err
+	}
+
+	ctx, err = events.WithContext(ctx, clientset)
 	if err != nil {
 		return err
 	}
@@ -91,7 +97,7 @@ func serverCommand(ctx context.Context, args *ServerCmd) (err error) {
 
 	eventsRecorder.Initializing()
 
-	r, err := routing.NewRouter(ctx, args.RouterAddr, httpsPort)
+	r, err := routing.NewRouter(ctx, clientset, args.RouterAddr, httpsPort)
 	if err != nil {
 		return err
 	}

go.mod

@@ -18,8 +18,8 @@ require (
 	github.com/schollz/progressbar/v3 v3.14.2
 	github.com/swaggo/swag v1.16.3
 	golang.org/x/sync v0.6.0
-	k8s.io/api v0.29.2
-	k8s.io/apimachinery v0.29.2
+	k8s.io/api v0.29.3
+	k8s.io/apimachinery v0.29.3
 )
 
 require (
@@ -49,6 +49,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/elastic/gosigar v0.14.2 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.3 // indirect
+	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/flynn/noise v1.1.0 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
@@ -69,7 +70,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/glog v1.2.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
@@ -205,6 +206,6 @@ require (
 	golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
-	k8s.io/client-go v0.29.2
+	k8s.io/client-go v0.29.3
 	lukechampine.com/blake3 v1.2.1 // indirect
 )

go.sum

@@ -207,6 +207,8 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
 github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/gnostic-models v0.6.8 h1:yo/ABAfM5IMRsS1VnXjTBvUb61tFIHozhlYvRgGre9I=
 github.com/google/gnostic-models v0.6.8/go.mod h1:5n7qKqH0f5wFt+aWF8CW6pZLLNOfYuF5OpfBSENuI8U=
@@ -848,10 +850,16 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 k8s.io/api v0.29.2 h1:hBC7B9+MU+ptchxEqTNW2DkUosJpp1P+Wn6YncZ474A=
 k8s.io/api v0.29.2/go.mod h1:sdIaaKuU7P44aoyyLlikSLayT6Vb7bvJNCX105xZXY0=
+k8s.io/api v0.29.3 h1:2ORfZ7+bGC3YJqGpV0KSDDEVf8hdGQ6A03/50vj8pmw=
+k8s.io/api v0.29.3/go.mod h1:y2yg2NTyHUUkIoTC+phinTnEa3KFM6RZ3szxt014a80=
 k8s.io/apimachinery v0.29.2 h1:EWGpfJ856oj11C52NRCHuU7rFDwxev48z+6DSlGNsV8=
 k8s.io/apimachinery v0.29.2/go.mod h1:6HVkd1FwxIagpYrHSwJlQqZI3G9LfYWRPAkUvLnXTKU=
+k8s.io/apimachinery v0.29.3 h1:2tbx+5L7RNvqJjn7RIuIKu9XTsIZ9Z5wX2G22XAa5EU=
+k8s.io/apimachinery v0.29.3/go.mod h1:hx/S4V2PNW4OMg3WizRrHutyB5la0iCUbZym+W0EQIU=
 k8s.io/client-go v0.29.2 h1:FEg85el1TeZp+/vYJM7hkDlSTFZ+c5nnK44DJ4FyoRg=
 k8s.io/client-go v0.29.2/go.mod h1:knlvFZE58VpqbQpJNbCbctTVXcd35mMyAAwBdpt4jrA=
+k8s.io/client-go v0.29.3 h1:R/zaZbEAxqComZ9FHeQwOh3Y1ZUs7FaHKZdQtIc2WZg=
+k8s.io/client-go v0.29.3/go.mod h1:tkDisCvgPfiRpxGnOORfkljmS+UrW+WtXAy2fTvXJB0=
 k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
 k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 h1:BZqlfIlq5YbRMFko6/PM7FjZpUb45WallggurYhKGag=

internal/k8s/events/events.go

@@ -11,7 +11,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	typedv1core "k8s.io/client-go/kubernetes/typed/core/v1"
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
 )
 
@@ -22,17 +21,9 @@ var (
 )
 
 // NewRecorder creates a new event recorder.
-func NewRecorder(ctx context.Context) (EventRecorder, error) {
-	clientset, err := k8s.NewKubernetesInterface(p2pcontext.KubeConfigPath)
-	if err != nil {
-		return nil, err
-	}
-
-	inPod := false
-	_, err = rest.InClusterConfig() // Assume run in a Pod or an environment with appropriate env variables set
-	if err == nil {
-		inPod = true
-	}
+func NewRecorder(ctx context.Context, k *k8s.ClientSet) (EventRecorder, error) {
+	clientset := k.Interface
+	inPod := k.InPod
 
 	var objRef *v1.ObjectReference
 	if !inPod {
@@ -74,8 +65,8 @@ func NewRecorder(ctx context.Context) (EventRecorder, error) {
 }
 
 // WithContext returns a new context with an event recorder.
-func WithContext(ctx context.Context) (context.Context, error) {
-	er, err := NewRecorder(ctx)
+func WithContext(ctx context.Context, clientset *k8s.ClientSet) (context.Context, error) {
+	er, err := NewRecorder(ctx, clientset)
 	if err != nil {
 		return nil, err
 	}

internal/k8s/events/events_test.go

@@ -6,11 +6,113 @@ import (
 	"context"
 	"testing"
 
+	p2pcontext "github.com/azure/peerd/internal/context"
+	"github.com/azure/peerd/pkg/k8s"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/record"
 )
 
+func TestWithContext(t *testing.T) {
+	fcs := fake.NewSimpleClientset([]runtime.Object{
+		&v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      p2pcontext.NodeName,
+				Namespace: p2pcontext.Namespace,
+				UID:       "test-uid",
+			},
+		},
+	}...)
+
+	cs := &k8s.ClientSet{Interface: fcs, InPod: true}
+
+	ctx, err := WithContext(context.Background(), cs)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if ctx == nil {
+		t.Fatal("expected context")
+	}
+
+	er := FromContext(ctx).(*eventRecorder)
+	if er.objRef.Kind != "Pod" {
+		t.Errorf("expected kind to be Pod, got %s", er.objRef.Kind)
+	}
+}
+
+func TestNewRecorderInNode(t *testing.T) {
+	fcs := fake.NewSimpleClientset([]runtime.Object{
+		&v1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: p2pcontext.NodeName,
+				UID:  "test-uid",
+			},
+		},
+	}...)
+
+	cs := &k8s.ClientSet{Interface: fcs, InPod: false}
+
+	r, err := NewRecorder(context.Background(), cs)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if r == nil {
+		t.Fatal("expected event recorder")
+	}
+
+	er := r.(*eventRecorder)
+	if er.objRef.Kind != "Node" {
+		t.Errorf("expected kind to be Node, got %s", er.objRef.Kind)
+	}
+	if er.objRef.Name != p2pcontext.NodeName {
+		t.Errorf("expected name to be %s, got %s", p2pcontext.NodeName, er.objRef.Name)
+	}
+	if er.objRef.UID != "test-uid" {
+		t.Errorf("expected uid to be test-uid, got %s", er.objRef.UID)
+	}
+}
+
+func TestNewRecorderInPod(t *testing.T) {
+	fcs := fake.NewSimpleClientset([]runtime.Object{
+		&v1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      p2pcontext.NodeName,
+				Namespace: p2pcontext.Namespace,
+				UID:       "test-uid",
+			},
+		},
+	}...)
+
+	cs := &k8s.ClientSet{Interface: fcs, InPod: true}
+
+	r, err := NewRecorder(context.Background(), cs)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if r == nil {
+		t.Fatal("expected event recorder")
+	}
+
+	er := r.(*eventRecorder)
+	if er.objRef.Kind != "Pod" {
+		t.Errorf("expected kind to be Pod, got %s", er.objRef.Kind)
+	}
+	if er.objRef.Name != p2pcontext.NodeName {
+		t.Errorf("expected name to be %s, got %s", p2pcontext.NodeName, er.objRef.Name)
+	}
+	if er.objRef.Namespace != p2pcontext.Namespace {
+		t.Errorf("expected namespace to be %s, got %s", p2pcontext.Namespace, er.objRef.Namespace)
+	}
+	if er.objRef.UID != "test-uid" {
+		t.Errorf("expected uid to be test-uid, got %s", er.objRef.UID)
+	}
+}
+
 func TestExpectedEvents(t *testing.T) {
 	er := &eventRecorder{
 		recorder: &testRecorder{t},

internal/routing/router.go

@@ -10,6 +10,7 @@ import (
 
 	p2pcontext "github.com/azure/peerd/internal/context"
 	"github.com/azure/peerd/internal/k8s/events"
+	"github.com/azure/peerd/pkg/k8s"
 	"github.com/azure/peerd/pkg/k8s/election"
 	"github.com/azure/peerd/pkg/peernet"
 	"github.com/dgraph-io/ristretto"
@@ -26,6 +27,7 @@ import (
 )
 
 type router struct {
+	clientset   *k8s.ClientSet
 	p2pnet      peernet.Network
 	host        host.Host
 	rd          *routing.RoutingDiscovery
@@ -42,7 +44,7 @@ type PeerNotFoundError struct {
 }
 
 // NewRouter creates a new router.
-func NewRouter(ctx context.Context, routerAddr, serverPort string) (Router, error) {
+func NewRouter(ctx context.Context, clientset *k8s.ClientSet, routerAddr, serverPort string) (Router, error) {
 	log := zerolog.Ctx(ctx).With().Str("component", "router").Logger()
 
 	h, p, err := net.SplitHostPort(routerAddr)
@@ -137,6 +139,7 @@ func NewRouter(ctx context.Context, routerAddr, serverPort string) (Router, erro
 	}
 
 	return &router{
+		clientset:   clientset,
 		p2pnet:      n,
 		host:        host,
 		rd:          rd,
@@ -196,7 +199,7 @@ func (r *router) Resolve(ctx context.Context, key string, allowSelf bool, count
 			peerCh <- PeerInfo{info.ID, fmt.Sprintf("https://%s:%s", v, r.port)}
 
 			if r.active.CompareAndSwap(false, true) {
-				er, err := events.NewRecorder(ctx)
+				er, err := events.NewRecorder(ctx, r.clientset)
 				if err != nil {
 					log.Error().Err(err).Msg("could not create event recorder")
 				} else {

internal/routing/router_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	p2pcontext "github.com/azure/peerd/internal/context"
+	"github.com/azure/peerd/pkg/k8s"
 	"github.com/dgraph-io/ristretto"
 	cid "github.com/ipfs/go-cid"
 	"github.com/libp2p/go-libp2p/core/connmgr"
@@ -21,8 +22,11 @@ import (
 	corerouting "github.com/libp2p/go-libp2p/core/routing"
 	"github.com/libp2p/go-libp2p/p2p/discovery/routing"
 	multiaddr "github.com/multiformats/go-multiaddr"
+	"k8s.io/client-go/kubernetes/fake"
 )
 
+var fakeClientset = k8s.ClientSet{Interface: fake.NewSimpleClientset(), InPod: true}
+
 func TestResolveWithCache(t *testing.T) {
 	c, err := ristretto.NewCache(&ristretto.Config{
 		NumCounters: 1e7,
@@ -41,6 +45,7 @@ func TestResolveWithCache(t *testing.T) {
 	}
 
 	r := &router{
+		clientset:   &fakeClientset,
 		host:        h,
 		port:        "5000",
 		lookupCache: c,
@@ -79,6 +84,7 @@ func TestResolve(t *testing.T) {
 	}
 
 	r := &router{
+		clientset:   &fakeClientset,
 		host:        h,
 		port:        "5000",
 		lookupCache: c,
@@ -134,6 +140,7 @@ func TestProvide(t *testing.T) {
 	}
 
 	r := &router{
+		clientset:   &fakeClientset,
 		host:        h,
 		port:        "5000",
 		lookupCache: c,

pkg/k8s/k8s.go

@@ -8,22 +8,37 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 )
 
+// ClientSet is an interface for k8s API server.
+type ClientSet struct {
+	kubernetes.Interface
+
+	// InPod indicates whether the current runtime environment is a pod.
+	InPod bool
+}
+
 // NewKubernetesInterface creates a new interface for k8s API server.
 // The current runtime environment is first assumed to be a pod and its identity is used to create the interface.
 // If a pod is not detected, the given kubeConfigPath is used to create the interface.
-func NewKubernetesInterface(kubeConfigPath string) (kubernetes.Interface, error) {
+func NewKubernetesInterface(kubeConfigPath string) (*ClientSet, error) {
+	k := &ClientSet{}
+
 	config, err := rest.InClusterConfig() // Assume run in a Pod or an environment with appropriate env variables set.
 	if err != nil {
 		config, err = clientcmd.BuildConfigFromFlags("", kubeConfigPath)
 		if err != nil {
 			return nil, err
 		}
+		k.InPod = false
+	} else {
+		k.InPod = true
 	}
 
 	clientset, err := kubernetes.NewForConfig(config)
 	if err != nil {
 		return nil, err
 	}
 
-	return clientset, nil
+	k.Interface = clientset
+
+	return k, nil
 }