Extend Machine Runtime Interface to Support Cross-Cluster Events (#1081)

broker/machinebroker/apiutils/apiutils.go

@@ -167,3 +167,19 @@ func IsManagedBy(o metav1.Object, manager string) bool {
 	actual, ok := o.GetLabels()[machinebrokerv1alpha1.ManagerLabel]
 	return ok && actual == manager
 }
+
+func GetIRIObjectMetadata(o metav1.Object) *irimeta.ObjectMetadata {
+	var deletedAt int64
+	if !o.GetDeletionTimestamp().IsZero() {
+		deletedAt = o.GetDeletionTimestamp().UnixNano()
+	}
+
+	return &irimeta.ObjectMetadata{
+		Id:          o.GetName(),
+		Annotations: o.GetAnnotations(),
+		Labels:      o.GetLabels(),
+		Generation:  o.GetGeneration(),
+		CreatedAt:   o.GetCreationTimestamp().UnixNano(),
+		DeletedAt:   deletedAt,
+	}
+}

broker/machinebroker/server/event_list.go

@@ -0,0 +1,93 @@
+// SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company and IronCore contributors
+// SPDX-License-Identifier: Apache-2.0
+
+package server
+
+import (
+	"context"
+	"fmt"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/ironcore-dev/ironcore/broker/machinebroker/apiutils"
+	iri "github.com/ironcore-dev/ironcore/iri/apis/machine/v1alpha1"
+)
+
+const InvolvedObjectName = "involvedObject.name"
+
+func (s *Server) listEvents(ctx context.Context, machineID string) ([]*iri.Event, error) {
+	machineEventList := &v1.EventList{}
+	selectorField := fields.Set{}
+	selectorField[InvolvedObjectName] = machineID
+
+	if err := s.cluster.Client().List(ctx, machineEventList,
+		client.InNamespace(s.cluster.Namespace()), client.MatchingFieldsSelector{Selector: selectorField.AsSelector()},
+	); err != nil {
+		return nil, fmt.Errorf("error listing machine events: %w", err)
+	}
+
+	var iriEvents []*iri.Event
+	for _, machineEvent := range machineEventList.Items {
+		iriEvent := &iri.Event{
+			Metadata: apiutils.GetIRIObjectMetadata(&machineEvent.ObjectMeta),
+			Spec: &iri.EventSpec{
+				Reason:    machineEvent.Reason,
+				Message:   machineEvent.Message,
+				Type:      machineEvent.Type,
+				EventTime: machineEvent.FirstTimestamp.Unix(),
+			},
+		}
+		iriEvents = append(iriEvents, iriEvent)
+	}
+	return iriEvents, nil
+}
+
+func (s *Server) filterEvents(events []*iri.Event, filter *iri.EventFilter) []*iri.Event {
+	if filter == nil {
+		return events
+	}
+
+	var (
+		res []*iri.Event
+		sel = labels.SelectorFromSet(filter.LabelSelector)
+	)
+	for _, iriEvent := range events {
+		if !sel.Matches(labels.Set(iriEvent.Metadata.Labels)) {
+			continue
+		}
+		if (filter.EventsFromTime >= 0 && iriEvent.Spec.EventTime >= filter.EventsFromTime) && (filter.EventsToTime >= 0 && iriEvent.Spec.EventTime <= filter.EventsToTime) {
+			res = append(res, iriEvent)
+		}
+
+	}
+	return res
+}
+
+func (s *Server) ListEvents(ctx context.Context, req *iri.ListEventsRequest) (*iri.ListEventsResponse, error) {
+	ironcoreMachineList, err := s.listIroncoreMachines(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var machineEvents []*iri.MachineEvents
+	for i := range ironcoreMachineList.Items {
+		ironcoreMachine := &ironcoreMachineList.Items[i]
+		iriEvents, err := s.listEvents(ctx, ironcoreMachine.Name)
+		if err != nil {
+			return nil, err
+		}
+
+		iriEvents = s.filterEvents(iriEvents, req.Filter)
+		ironcoreMachineMeta, err := apiutils.GetObjectMetadata(&ironcoreMachine.ObjectMeta)
+		if err != nil {
+			return nil, fmt.Errorf("error listing machine events: %w", err)
+		}
+		machineEvents = append(machineEvents, &iri.MachineEvents{InvolvedObjectMeta: ironcoreMachineMeta, Events: iriEvents})
+	}
+
+	return &iri.ListEventsResponse{
+		MachineEvents: machineEvents,
+	}, nil
+}

broker/machinebroker/server/event_list_test.go

@@ -0,0 +1,86 @@
+// SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company and IronCore contributors
+// SPDX-License-Identifier: Apache-2.0
+
+package server_test
+
+import (
+	computev1alpha1 "github.com/ironcore-dev/ironcore/api/compute/v1alpha1"
+	networkingv1alpha1 "github.com/ironcore-dev/ironcore/api/networking/v1alpha1"
+	iri "github.com/ironcore-dev/ironcore/iri/apis/machine/v1alpha1"
+	irimeta "github.com/ironcore-dev/ironcore/iri/apis/meta/v1alpha1"
+	machinepoolletv1alpha1 "github.com/ironcore-dev/ironcore/poollet/machinepoollet/api/v1alpha1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+var _ = Describe("ListEvents", func() {
+	ns, srv := SetupTest()
+	machineClass := SetupMachineClass()
+
+	It("should correctly list events", func(ctx SpecContext) {
+		By("creating machine")
+		Expect(computev1alpha1.AddToScheme(scheme.Scheme)).To(Succeed())
+		Expect(networkingv1alpha1.AddToScheme(scheme.Scheme)).To(Succeed())
+
+		k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
+			Scheme: scheme.Scheme,
+		})
+		Expect(err).ToNot(HaveOccurred())
+
+		res, err := srv.CreateMachine(ctx, &iri.CreateMachineRequest{
+			Machine: &iri.Machine{
+				Metadata: &irimeta.ObjectMetadata{
+					Labels: map[string]string{
+						machinepoolletv1alpha1.MachineUIDLabel: "foobar",
+					},
+				},
+				Spec: &iri.MachineSpec{
+					Power: iri.Power_POWER_ON,
+					Image: &iri.ImageSpec{
+						Image: "example.org/foo:latest",
+					},
+					Class: machineClass.Name,
+					NetworkInterfaces: []*iri.NetworkInterface{
+						{
+							Name:      "primary-nic",
+							NetworkId: "network-id",
+							Ips:       []string{"10.0.0.1"},
+						},
+					},
+				},
+			},
+		})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(res).NotTo(BeNil())
+
+		By("getting the ironcore machine")
+		ironcoreMachine := &computev1alpha1.Machine{}
+		ironcoreMachineKey := client.ObjectKey{Namespace: ns.Name, Name: res.Machine.Metadata.Id}
+		Expect(k8sClient.Get(ctx, ironcoreMachineKey, ironcoreMachine)).To(Succeed())
+
+		By("generating the machine events")
+		eventRecorder := k8sManager.GetEventRecorderFor("test-recorder")
+		eventRecorder.Event(ironcoreMachine, corev1.EventTypeNormal, "testing", "this is test event")
+
+		By("listing the machine events")
+		resp, err := srv.ListEvents(ctx, &iri.ListEventsRequest{})
+
+		Expect(err).NotTo(HaveOccurred())
+
+		Expect(resp.MachineEvents).To(ConsistOf(SatisfyAll(
+			HaveField("InvolvedObjectMeta.Id", Equal(ironcoreMachine.Name)),
+			HaveField("Events", ConsistOf(SatisfyAll(
+				HaveField("Spec", SatisfyAll(
+					HaveField("Reason", Equal("testing")),
+					HaveField("Message", Equal("this is test event")),
+					HaveField("Type", Equal(corev1.EventTypeNormal)),
+				)),
+			))),
+		)))
+	})
+})

broker/machinebroker/server/machine_list.go

@@ -23,14 +23,8 @@ import (
 )
 
 func (s *Server) listAggregateIronCoreMachines(ctx context.Context) ([]AggregateIronCoreMachine, error) {
-	ironcoreMachineList := &computev1alpha1.MachineList{}
-	if err := s.cluster.Client().List(ctx, ironcoreMachineList,
-		client.InNamespace(s.cluster.Namespace()),
-		client.MatchingLabels{
-			machinebrokerv1alpha1.ManagerLabel: machinebrokerv1alpha1.MachineBrokerManager,
-			machinebrokerv1alpha1.CreatedLabel: "true",
-		},
-	); err != nil {
+	ironcoreMachineList, err := s.listIroncoreMachines(ctx)
+	if err != nil {
 		return nil, fmt.Errorf("error listing ironcore machines: %w", err)
 	}
 
@@ -61,6 +55,20 @@ func (s *Server) listAggregateIronCoreMachines(ctx context.Context) ([]Aggregate
 	return res, nil
 }
 
+func (s *Server) listIroncoreMachines(ctx context.Context) (*computev1alpha1.MachineList, error) {
+	ironcoreMachineList := &computev1alpha1.MachineList{}
+	if err := s.cluster.Client().List(ctx, ironcoreMachineList,
+		client.InNamespace(s.cluster.Namespace()),
+		client.MatchingLabels{
+			machinebrokerv1alpha1.ManagerLabel: machinebrokerv1alpha1.MachineBrokerManager,
+			machinebrokerv1alpha1.CreatedLabel: "true",
+		},
+	); err != nil {
+		return nil, err
+	}
+	return ironcoreMachineList, nil
+}
+
 func (s *Server) aggregateIronCoreMachine(
 	ctx context.Context,
 	rd client.Reader,

broker/machinebroker/server/machine_networkinterface_attach.go

@@ -100,7 +100,6 @@ func (s *Server) createIronCoreNetworkInterface(
 
 func (s *Server) attachIronCoreNetworkInterface(
 	ctx context.Context,
-	log logr.Logger,
 	ironcoreMachine *computev1alpha1.Machine,
 	ironcoreMachineNic *computev1alpha1.NetworkInterface,
 ) error {
@@ -141,7 +140,7 @@ func (s *Server) AttachNetworkInterface(ctx context.Context, req *iri.AttachNetw
 		return nil, err
 	}
 
-	if err := s.attachIronCoreNetworkInterface(ctx, log, ironcoreMachine, ironcoreMachineNic); err != nil {
+	if err := s.attachIronCoreNetworkInterface(ctx, ironcoreMachine, ironcoreMachineNic); err != nil {
 		return nil, fmt.Errorf("error creating ironcore network interface: %w", err)
 	}
 

broker/machinebroker/server/machine_volume_attach.go

@@ -212,7 +212,6 @@ func (s *Server) createIronCoreVolume(
 
 func (s *Server) attachIronCoreVolume(
 	ctx context.Context,
-	log logr.Logger,
 	ironcoreMachine *computev1alpha1.Machine,
 	ironcoreMachineVolume *computev1alpha1.Volume,
 ) error {
@@ -251,7 +250,7 @@ func (s *Server) AttachVolume(ctx context.Context, req *iri.AttachVolumeRequest)
 	}
 
 	log.V(1).Info("Attaching ironcore volume")
-	if err := s.attachIronCoreVolume(ctx, log, ironcoreMachine, ironcoreMachineVolume); err != nil {
+	if err := s.attachIronCoreVolume(ctx, ironcoreMachine, ironcoreMachineVolume); err != nil {
 		return nil, err
 	}
 

iri/apis/machine/machine.go

@@ -11,6 +11,7 @@ import (
 
 type RuntimeService interface {
 	Version(context.Context, *api.VersionRequest) (*api.VersionResponse, error)
+	ListEvents(context.Context, *api.ListEventsRequest) (*api.ListEventsResponse, error)
 	ListMachines(context.Context, *api.ListMachinesRequest) (*api.ListMachinesResponse, error)
 	CreateMachine(context.Context, *api.CreateMachineRequest) (*api.CreateMachineResponse, error)
 	DeleteMachine(context.Context, *api.DeleteMachineRequest) (*api.DeleteMachineResponse, error)