Record ACI events back to kubernetes (virtual-kubelet#470)

Signed-off-by: JACQUES Francois <Francois.JACQUES@murex.com>

cmd/virtual-kubelet/main.go

@@ -198,13 +198,12 @@ func main() {
 		}
 		return nil
 	}
-
+	k8sClient, err := nodeutil.ClientsetFromEnv(kubeConfigPath)
+	if err != nil {
+		log.G(ctx).Fatal(err)
+	}
 	withClient := func(cfg *nodeutil.NodeConfig) error {
-		client, err := nodeutil.ClientsetFromEnv(kubeConfigPath)
-		if err != nil {
-			return err
-		}
-		return nodeutil.WithClient(client)(cfg)
+		return nodeutil.WithClient(k8sClient)(cfg)
 	}
 
 	run := func(ctx context.Context) error {
@@ -223,7 +222,7 @@ func main() {
 				}
 				p, err := azproviderv2.NewACIProvider(ctx, cfgPath, azConfig, azACIAPIs, cfg,
 					nodeName, operatingSystem, os.Getenv("VKUBELET_POD_IP"),
-					int32(listenPort), clusterDomain)
+					int32(listenPort), clusterDomain, k8sClient)
 				p.ConfigureNode(ctx, cfg.Node)
 				return p, nil, err
 			},

pkg/featureflag/feature_flag.go

@@ -13,11 +13,15 @@ import (
 const (
 	InitContainerFeature       = "init-container"
 	ConfidentialComputeFeature = "confidential-compute"
+
+	// Events : support ACI to K8s event translation and broadcasting
+	Events = "events"
 )
 
 var enabledFeatures = []string{
 	InitContainerFeature,
 	ConfidentialComputeFeature,
+	Events,
 }
 
 type FlagIdentifier struct {
@@ -30,6 +34,8 @@ func InitFeatureFlag(ctx context.Context) *FlagIdentifier {
 	var featureFlags FlagIdentifier
 	featureFlags.enabledFeatures = enabledFeatures
 
+	log.G(ctx).Infof("features %v enabled", enabledFeatures)
+
 	return &featureFlags
 }
 
@@ -42,9 +48,10 @@ func (fi *FlagIdentifier) IsEnabled(ctx context.Context, feature string) bool {
 	}
 	for _, feat := range fi.enabledFeatures {
 		if feat == feature {
-			log.G(ctx).Infof("feature %s is enabled", feature)
+			log.G(ctx).Debugf("feature %s is enabled", feature)
 			return true
 		}
 	}
+	log.G(ctx).Debugf("feature %s is disabled", feature)
 	return false
 }

pkg/featureflag/feature_flag_test.go

@@ -29,6 +29,11 @@ func TestIsEnabled(t *testing.T) {
 			feature:       ConfidentialComputeFeature,
 			shouldEnabled: true,
 		},
+		{
+			description:   fmt.Sprintf(" %s feature should be enabled", Events),
+			feature:       Events,
+			shouldEnabled: true,
+		},
 	}
 	for i, tc := range cases {
 		t.Run(tc.description, func(t *testing.T) {

pkg/provider/aci.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path"
 	"reflect"
 	"strings"
 	"time"
@@ -33,10 +34,16 @@ import (
 	"github.com/virtual-kubelet/virtual-kubelet/trace"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 
 	"github.com/cpuguy83/dockercfg"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/tools/reference"
 )
 
 const (
@@ -80,6 +87,7 @@ type ACIProvider struct {
 	podsL                    corev1listers.PodLister
 	enabledFeatures          *featureflag.FlagIdentifier
 	providerNetwork          network.ProviderNetwork
+	eventRecorder            record.EventRecorder
 
 	resourceGroup      string
 	region             string
@@ -170,7 +178,7 @@ func isValidACIRegion(region string) bool {
 }
 
 // NewACIProvider creates a new ACIProvider.
-func NewACIProvider(ctx context.Context, config string, azConfig auth.Config, azAPIs client.AzClientsInterface, pCfg nodeutil.ProviderConfig, nodeName, operatingSystem string, internalIP string, daemonEndpointPort int32, clusterDomain string) (*ACIProvider, error) {
+func NewACIProvider(ctx context.Context, config string, azConfig auth.Config, azAPIs client.AzClientsInterface, pCfg nodeutil.ProviderConfig, nodeName, operatingSystem string, internalIP string, daemonEndpointPort int32, clusterDomain string, kubeClient kubernetes.Interface) (*ACIProvider, error) {
 	var p ACIProvider
 	var err error
 
@@ -208,6 +216,12 @@ func NewACIProvider(ctx context.Context, config string, azConfig auth.Config, az
 	if p.providerNetwork.VnetResourceGroup == "" {
 		p.providerNetwork.VnetResourceGroup = p.resourceGroup
 	}
+
+	providerEB := record.NewBroadcaster()
+	providerEB.StartLogging(log.G(ctx).Infof)
+	providerEB.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: kubeClient.CoreV1().Events(v1.NamespaceAll)})
+	p.eventRecorder = providerEB.NewRecorder(scheme.Scheme, v1.EventSource{Component: path.Join(nodeName, "pod-controller")})
+
 	// If the log analytics file has been specified, load workspace credentials from the file
 	if logAnalyticsAuthFile := os.Getenv("LOG_ANALYTICS_AUTH_LOCATION"); logAnalyticsAuthFile != "" {
 		p.diagnostics, err = analytics.NewContainerGroupDiagnosticsFromFile(logAnalyticsAuthFile)
@@ -761,6 +775,8 @@ func (p *ACIProvider) NotifyPods(ctx context.Context, notifierCb func(*v1.Pod))
 		pods:     p.podsL,
 		updateCb: notifierCb,
 		handler:  p,
+		lastEventCheck: time.UnixMicro(0),
+		eventRecorder:  p.eventRecorder,
 	}
 
 	go p.tracker.StartTracking(ctx)
@@ -797,6 +813,46 @@ func (p *ACIProvider) FetchPodStatus(ctx context.Context, ns, name string) (*v1.
 	return p.GetPodStatus(ctx, ns, name)
 }
 
+func (p *ACIProvider) FetchPodEvents(ctx context.Context, pod *v1.Pod, evtSink func(timestamp *time.Time, object runtime.Object, eventtype, reason, messageFmt string, args ...interface{})) error {
+	ctx, span := trace.StartSpan(ctx, "ACIProvider.FetchPodEvents")
+	defer span.End()
+	if !p.enabledFeatures.IsEnabled(ctx, featureflag.Events) {
+		return nil
+	}
+
+	ctx = addAzureAttributes(ctx, span, p)
+	cgName := containerGroupName(pod.Namespace, pod.Name)
+	cg, err := p.azClientsAPIs.GetContainerGroup(ctx, p.resourceGroup, cgName)
+	if err != nil {
+		return err
+	}
+	if *cg.Tags["NodeName"] != p.nodeName {
+		return errors.Wrapf(err, "container group %s found with mismatching node", cgName)
+	}
+
+	if cg.Properties != nil && cg.Properties.InstanceView != nil && cg.Properties.InstanceView.Events != nil {
+		for _, evt := range cg.Properties.InstanceView.Events {
+			evtSink(evt.LastTimestamp, pod, *evt.Type, *evt.Name, *evt.Message)
+		}
+	}
+
+	if cg.Properties != nil && cg.Properties.Containers != nil {
+		for _, container := range cg.Properties.Containers {
+			if container.Properties != nil && container.Properties.InstanceView != nil && container.Properties.InstanceView.Events != nil {
+				for _, evt := range container.Properties.InstanceView.Events {
+					podReference, err := reference.GetReference(scheme.Scheme, pod)
+					if err != nil {
+						log.G(ctx).WithError(err).Warnf("cannot get k8s object reference from pod %s in namespace %s", pod.Name, pod.Namespace)
+					}
+					podReference.FieldPath = fmt.Sprintf("spec.containers{%s}", *container.Name)
+					evtSink(evt.LastTimestamp, podReference, *evt.Type, *evt.Name, *evt.Message)
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // CleanupPod interface impl
 func (p *ACIProvider) CleanupPod(ctx context.Context, ns, name string) error {
 	ctx, span := trace.StartSpan(ctx, "ACIProvider.CleanupPod")

pkg/provider/aci_confidential_test.go

@@ -123,7 +123,7 @@ func TestCreatePodWithConfidentialComputeProperties(t *testing.T) {
 			ctx := context.TODO()
 
 			provider, err := createTestProvider(aciMocks, NewMockConfigMapLister(mockCtrl),
-				NewMockSecretLister(mockCtrl), NewMockPodLister(mockCtrl))
+				NewMockSecretLister(mockCtrl), NewMockPodLister(mockCtrl), nil)
 			if err != nil {
 				t.Fatal("Unable to create test provider", err)
 			}

pkg/provider/aci_init_container_test.go

@@ -183,7 +183,7 @@ func TestCreatePodWithInitContainers(t *testing.T) {
 			ctx := context.TODO()
 
 			provider, err := createTestProvider(aciMocks, NewMockConfigMapLister(mockCtrl),
-				NewMockSecretLister(mockCtrl), NewMockPodLister(mockCtrl))
+				NewMockSecretLister(mockCtrl), NewMockPodLister(mockCtrl), nil)
 			if err != nil {
 				t.Fatal("Unable to create test provider", err)
 			}