Fix #46

kubernetes.container_image and docker.container_id are now attached to
mounted-file originating logs.

Author: jvassev

config-reloader/datasource/datasource.go

@@ -5,6 +5,7 @@ package datasource
 
 import (
 	"sort"
+
 	core "k8s.io/api/core/v1"
 )
 
@@ -23,6 +24,9 @@ type MiniContainer struct {
 	PodID   string
 	PodName string
 
+	Image       string
+	ContainerID string
+
 	// pod labels
 	Labels map[string]string
 
@@ -71,17 +75,34 @@ func (s byLength) Less(i, j int) bool {
 	return len(s[i].Path) > len(s[j].Path)
 }
 
+func findContainerStatus(statuses []core.ContainerStatus, name string) *core.ContainerStatus {
+	for _, st := range statuses {
+		if st.Name == name {
+			return &st
+		}
+	}
+	return nil
+}
+
 func convertPodToMinis(resp *core.PodList) []*MiniContainer {
 	var res []*MiniContainer
 
 	for _, pod := range resp.Items {
 		for _, cont := range pod.Spec.Containers {
+			contStatus := findContainerStatus(pod.Status.ContainerStatuses, cont.Name)
+			cid := ""
+			if contStatus != nil {
+				cid = contStatus.ContainerID
+			}
+
 			mini := &MiniContainer{
-				PodID:    string(pod.UID),
-				PodName:  pod.Name,
-				Labels:   pod.Labels,
-				Name:     cont.Name,
-				NodeName: pod.Spec.NodeName,
+				PodID:       string(pod.UID),
+				PodName:     pod.Name,
+				Labels:      pod.Labels,
+				Name:        cont.Name,
+				NodeName:    pod.Spec.NodeName,
+				Image:       cont.Image,
+				ContainerID: cid,
 			}
 
 			for _, vm := range cont.VolumeMounts {

config-reloader/processors/mounted_file.go

@@ -162,11 +162,16 @@ func (state *mountedFileState) makeAttachK8sMetadataDirective(tag string, mc *da
 	buf := &bytes.Buffer{}
 	fmt.Fprintf(buf, "record['stream']='%s'; ", cf.Path)
 	fmt.Fprintf(buf, "record['kubernetes']=%s; ", util.ToRubyMapLiteral(map[string]string{
-		"container_name": mc.Name,
-		"namespace_name": state.Context.Namepsace,
-		"pod_name":       mc.PodName,
-		"pod_id":         mc.PodID,
-		"host":           mc.NodeName,
+		"container_name":  mc.Name,
+		"container_image": mc.Image,
+		"namespace_name":  state.Context.Namepsace,
+		"pod_name":        mc.PodName,
+		"pod_id":          mc.PodID,
+		"host":            mc.NodeName,
+	}))
+
+	fmt.Fprintf(buf, "record['docker']=%s; ", util.ToRubyMapLiteral(map[string]string{
+		"container_id": mc.ContainerID,
 	}))
 
 	fmt.Fprintf(buf, "record['container_info']='%s'; ", util.Hash(mc.PodID, cf.Path))

config-reloader/processors/mounted_file_test.go

@@ -66,7 +66,7 @@ func TestMountedFileCatchesMissingFile(t *testing.T) {
 	  @type mounted-file
 	  labels app=spring-mvc
 	</source>
-	
+
 	<match **>
 	  @type logzio
 	</match
@@ -80,9 +80,9 @@ func TestMountedFileCatchesEmptyLabels(t *testing.T) {
 	missingPath := `
 	<source>
 	  @type mounted-file
-	  labels 
+	  labels
 	</source>
-	
+
 	<match **>
 	  @type logzio
 	</match
@@ -98,7 +98,7 @@ func TestMountedFileCatchesMissingLabels(t *testing.T) {
 	  @type mounted-file
 	  file /etc/hosts
 	</source>
-	
+
 	<match **>
 	  @type logzio
 	</match
@@ -235,10 +235,12 @@ func TestConvertToFragment(t *testing.T) {
 
 func TestProcessMountedFile(t *testing.T) {
 	c1 := &datasource.MiniContainer{
-		PodID:   "123-id",
-		PodName: "123",
-		Name:    "redis-main",
-		Labels:  map[string]string{"app": "redis"},
+		PodID:       "123-id",
+		PodName:     "123",
+		Image:       "image-c1",
+		ContainerID: "contid-c1",
+		Name:        "redis-main",
+		Labels:      map[string]string{"app": "redis"},
 		HostMounts: []*datasource.Mount{
 			{
 				Path:       "/var/log",
@@ -248,10 +250,12 @@ func TestProcessMountedFile(t *testing.T) {
 	}
 
 	c2 := &datasource.MiniContainer{
-		PodID:   "abc-id",
-		PodName: "abc",
-		Name:    "nginx-main",
-		Labels:  map[string]string{"app": "nginx"},
+		PodID:       "abc-id",
+		PodName:     "abc",
+		Image:       "image-c2",
+		ContainerID: "contid-c2",
+		Name:        "nginx-main",
+		Labels:      map[string]string{"app": "nginx"},
 		HostMounts: []*datasource.Mount{
 			{
 				Path:       "/var/log",
@@ -306,6 +310,12 @@ func TestProcessMountedFile(t *testing.T) {
 	assert.Equal(t, "/kubelet-root/pods/123-id/volumes/kubernetes.io~empty-dir/logs/redis.log", prep[0].Param("path"))
 	assert.Equal(t, "/kubelet-root/pods/abc-id/volumes/kubernetes.io~empty-dir/logs/nginx.log", prep[2].Param("path"))
 
+	payload := prep.String()
+	assert.True(t, strings.Contains(payload, "'container_image'=>'image-c2'"))
+	assert.True(t, strings.Contains(payload, "'container_image'=>'image-c1'"))
+	assert.True(t, strings.Contains(payload, "record['docker']={'container_id'=>'contid-c1'}"))
+	assert.True(t, strings.Contains(payload, "record['docker']={'container_id'=>'contid-c2'}"))
+
 	main, err := Process(input, ctx, state)
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(main))