Merge pull request #317 from intelops/multicontainer-imagename

imagename

agent/kubviz/k8smetrics_agent.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/intelops/go-common/logging"
+	"github.com/pkg/errors"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 
@@ -205,7 +206,7 @@ func publishMetrics(clientset *kubernetes.Clientset, js nats.JetStreamContext, e
 	errCh <- nil
 }
 
-func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStreamContext) (bool, error) {
+func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStreamContext, imageName string) (bool, error) {
 
 	ctx := context.Background()
 	tracer := otel.Tracer("kubviz-publish-k8smetrics")
@@ -218,6 +219,7 @@ func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStre
 		Type:        mtype,
 		Event:       mdata,
 		ClusterName: ClusterName,
+		ImageName: imageName,
 	}
 	metricsJson, _ := json.Marshal(metrics)
 	_, err := js.Publish(constants.EventSubject, metricsJson)
@@ -228,6 +230,27 @@ func publishK8sMetrics(id string, mtype string, mdata *v1.Event, js nats.JetStre
 	return false, nil
 }
 
+func getK8sPodImages(clientset *kubernetes.Clientset, namespace, podName string) ([]string, error) {
+    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+    defer cancel()
+
+    pod, err := clientset.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
+    if err != nil {
+        return nil, err
+    }
+
+    var images []string
+    for _, container := range pod.Spec.Containers {
+        images = append(images, container.Image)
+    }
+
+    if len(images) == 0 {
+        return nil, errors.New("no containers found in the pod")
+    }
+
+    return images, nil
+}
+
 // createStream creates a stream by using JetStreamContext
 func createStream(js nats.JetStreamContext) error {
 	// Check if the METRICS stream already exists; if not, create it.
@@ -327,15 +350,36 @@ func watchK8sEvents(clientset *kubernetes.Clientset, js nats.JetStreamContext) {
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				event := obj.(*v1.Event)
-				publishK8sMetrics(string(event.ObjectMeta.UID), "ADD", event, js)
+				images, err := getK8sPodImages(clientset, event.InvolvedObject.Namespace, event.InvolvedObject.Name)
+				if err != nil {
+                    log.Println("Error retrieving image names:", err)
+                    return
+                }
+				for _, image := range images {
+				publishK8sMetrics(string(event.ObjectMeta.UID), "ADD", event, js, image)
+				}
 			},
 			DeleteFunc: func(obj interface{}) {
 				event := obj.(*v1.Event)
-				publishK8sMetrics(string(event.ObjectMeta.UID), "DELETE", event, js)
+				images, err := getK8sPodImages(clientset, event.InvolvedObject.Namespace, event.InvolvedObject.Name)
+				if err != nil {
+                    log.Println("Error retrieving image names:", err)
+                    return
+                }
+                for _, image := range images {
+				publishK8sMetrics(string(event.ObjectMeta.UID), "DELETE", event, js, image)
+				}
 			},
 			UpdateFunc: func(oldObj, newObj interface{}) {
 				event := newObj.(*v1.Event)
-				publishK8sMetrics(string(event.ObjectMeta.UID), "UPDATE", event, js)
+				images, err := getK8sPodImages(clientset, event.InvolvedObject.Namespace, event.InvolvedObject.Name)
+                if err != nil {
+                    log.Println("Error retrieving image names:", err)
+                    return
+                }
+                for _, image := range images {
+				publishK8sMetrics(string(event.ObjectMeta.UID), "UPDATE", event, js, image)
+				}
 			},
 		},
 	)

client/pkg/clickhouse/db_client.go

@@ -543,6 +543,7 @@ func (c *DBClient) InsertKubvizEvent(metrics model.Metrics) {
 		metrics.Event.Reason,
 		metrics.Event.Source.Host,
 		string(eventJson),
+		metrics.ImageName,
 		formattedFirstTimestamp,
 		formattedLastTimestamp,
 	); err != nil {

client/pkg/clickhouse/statements.go

@@ -222,7 +222,7 @@ const InsertKetall DBStatement = "INSERT INTO getall_resources (ClusterName, Nam
 const InsertOutdated DBStatement = "INSERT INTO outdated_images (ClusterName, Namespace, Pod, CurrentImage, CurrentTag, LatestVersion, VersionsBehind, EventTime) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
 const InsertDepricatedApi DBStatement = "INSERT INTO DeprecatedAPIs (ClusterName, ObjectName, Description, Kind, Deprecated, Scope, EventTime) VALUES (?, ?, ?, ?, ?, ?, ?)"
 const InsertDeletedApi DBStatement = "INSERT INTO DeletedAPIs (ClusterName, ObjectName, Group, Kind, Version, Name, Deleted, Scope, EventTime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
-const InsertKubvizEvent DBStatement = "INSERT INTO events (ClusterName, Id, EventTime, OpType, Name, Namespace, Kind, Message, Reason, Host, Event, FirstTime, LastTime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+const InsertKubvizEvent DBStatement = "INSERT INTO events (ClusterName, Id, EventTime, OpType, Name, Namespace, Kind, Message, Reason, Host, Event, ImageName, FirstTime, LastTime) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
 const clickhouseExperimental DBStatement = `SET allow_experimental_object_type=1;`
 const containerGithubTable DBStatement = `CREATE table IF NOT EXISTS container_github(event JSON) ENGINE = MergeTree ORDER BY tuple();`
 const InsertKubeScore string = "INSERT INTO kubescore(id,clustername,object_name,kind,apiVersion,name,namespace,target_type,description,path,summary,file_name,file_row,EventTime) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"

model/metrics.go

@@ -7,4 +7,5 @@ type Metrics struct {
 	Type        string
 	Event       *v1.Event
 	ClusterName string
+	ImageName   string
 }

sql/000001_events.up.sql

@@ -10,6 +10,7 @@ CREATE TABLE IF NOT EXISTS events (
 	Reason      String,
 	Host        String,
 	Event       String,
+	ImageName   String,
 	FirstTime   String,
 	LastTime    String,
 	ExpiryDate DateTime DEFAULT now() + INTERVAL {{.TTLValue}} {{.TTLUnit}}