feat(model/message_handler): Add time to log message output

requires workflow-e2e#334
requires fluentd#61

Makefile

@@ -74,7 +74,7 @@ clean: check-docker
 	docker rmi $(IMAGE)
 
 update-manifests:
-	sed 's#\(image:\) .*#\1 $(IMAGE)#' manifests/deis-logger-rc.yaml > manifests/deis-logger-rc.tmp.yaml
+	sed 's#\(image:\) .*#\1 $(IMAGE)#' manifests/deis-logger-deployment.yaml > manifests/deis-logger-deployment.tmp.yaml
 
 test: test-style test-unit
 
@@ -122,19 +122,19 @@ test-unit: start-test-redis
 
 kube-install:
 	kubectl create -f manifests/deis-logger-svc.yaml
-	kubectl create -f manifests/deis-logger-rc.yaml
+	kubectl create -f manifests/deis-logger-deployment.yaml
 
 kube-delete:
 	-kubectl delete -f manifests/deis-logger-svc.yaml
-	-kubectl delete -f manifests/deis-logger-rc.tmp.yaml
+	-kubectl delete -f manifests/deis-logger-deployment.tmp.yaml
 
 kube-create: update-manifests
 	kubectl create -f manifests/deis-logger-svc.yaml
-	kubectl create -f manifests/deis-logger-rc.tmp.yaml
+	kubectl create -f manifests/deis-logger-deployment.tmp.yaml
 
 kube-replace: build push update-manifests
-	kubectl replace --force -f manifests/deis-logger-rc.tmp.yaml
+	kubectl replace --force -f manifests/deis-logger-deployment.tmp.yaml
 
 kube-update: update-manifests
-	kubectl delete -f manifests/deis-logger-rc.tmp.yaml
-	kubectl create -f manifests/deis-logger-rc.tmp.yaml
+	kubectl delete -f manifests/deis-logger-deployment.tmp.yaml
+	kubectl create -f manifests/deis-logger-deployment.tmp.yaml

glide.yaml

@@ -5,3 +5,7 @@ import:
 - package: github.com/kelseyhightower/envconfig
 - package: github.com/nsqio/go-nsq
 - package: gopkg.in/redis.v3
+- package: github.com/stretchr/testify
+  version: ~1.1.4
+  subpackages:
+  - assert

log/message_handler.go

@@ -4,17 +4,21 @@ import (
 	"encoding/json"
 	"fmt"
 	"regexp"
+	"strings"
 
 	"github.com/deis/logger/storage"
 )
 
 const (
+	podPattern              = `(\w.*)-(\w.*)-(\w.*)-(\w.*)`
 	controllerPattern       = `^(INFO|WARN|DEBUG|ERROR)\s+(\[(\S+)\])+:(.*)`
 	controllerContainerName = "deis-controller"
+	timeFormat              = "2006-01-02T15:04:05-07:00"
 )
 
 var (
-	regex = regexp.MustCompile(controllerPattern)
+	controllerRegex = regexp.MustCompile(controllerPattern)
+	podRegex        = regexp.MustCompile(podPattern)
 )
 
 func handle(rawMessage []byte, storageAdapter storage.Adapter) error {
@@ -23,7 +27,7 @@ func handle(rawMessage []byte, storageAdapter storage.Adapter) error {
 		return err
 	}
 	if fromController(message) {
-		storageAdapter.Write(getApplicationFromControllerMessage(message), message.Log)
+		storageAdapter.Write(getApplicationFromControllerMessage(message), buildControllerLogMessage(message))
 	} else {
 		labels := message.Kubernetes.Labels
 		storageAdapter.Write(labels["app"], buildApplicationLogMessage(message))
@@ -37,11 +41,23 @@ func fromController(message *Message) bool {
 }
 
 func getApplicationFromControllerMessage(message *Message) string {
-	return regex.FindStringSubmatch(message.Log)[3]
+	return controllerRegex.FindStringSubmatch(message.Log)[3]
+}
+
+func buildControllerLogMessage(message *Message) string {
+	l := controllerRegex.FindStringSubmatch(message.Log)
+	return fmt.Sprintf("%s deis[controller]: %s %s",
+		message.Time.Format(timeFormat),
+		l[1],
+		strings.Trim(l[4], " "))
 }
 
 func buildApplicationLogMessage(message *Message) string {
-	body := message.Log
-	podName := message.Kubernetes.PodName
-	return fmt.Sprintf("%s -- %s", podName, body)
+	return fmt.Sprintf("%s %s[%s.%s.%s]: %s",
+		message.Time.Format(timeFormat),
+		message.Kubernetes.Labels["app"],
+		message.Kubernetes.Labels["type"],
+		message.Kubernetes.Labels["version"],
+		podRegex.FindStringSubmatch(message.Kubernetes.PodName)[4],
+		message.Log)
 }

log/message_handler_test.go

@@ -0,0 +1,104 @@
+package log
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/deis/logger/storage"
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	validControllerMessage = `{"log": "INFO [foo]: admin deployed 2fd9226", "time": "2016-10-18T20:29:38+00:00", "stream": "stderr", "docker": {"container_id": "containerId"}, "kubernetes": {"namespace_name": "foo", "pod_id": "podId", "pod_name": "podName", "container_name": "deis-controller", "labels": {"app": "foo",
+"heritage": "deis", "type": "web", "version": "v2"}, "host": "host"}}`
+
+	invalidControllerMessage = `{"log": "INFO: admin deployed 2fd9226", "stream": "stderr", "docker": {"container_id": "containerId"}, "kubernetes": {"namespace_name": "foo", "pod_id": "podId", "pod_name": "podName", "container_name": "deis-controller", "labels": {"app": "foo",
+"heritage": "deis", "type": "web", "version": "v2"}, "host": "host"}}`
+
+	validAppMessage = `{"log": "test message", "stream": "stderr", "time": "2016-10-18T20:29:38+00:00", "docker": {"container_id": "containerId"}, "kubernetes": {"namespace_name": "foo", "pod_id": "podId", "pod_name": "foo-web-845861952-nzf60", "container_name": "foo-web", "labels": {"app": "foo",
+"heritage": "deis", "type": "web", "version": "v2"}, "host": "host"}}`
+
+	badjson = `{"log":}`
+)
+
+func TestValidControllerMessage(t *testing.T) {
+	message := new(Message)
+	err := json.Unmarshal([]byte(validControllerMessage), message)
+	assert.NoError(t, err, "error occured parsing log message")
+	assert.True(t, fromController(message), "json is not from controller")
+}
+
+func TestInvalidControllerMessage(t *testing.T) {
+	message := new(Message)
+	err := json.Unmarshal([]byte(validAppMessage), message)
+	assert.NoError(t, err, "error occured parsing log message")
+	assert.False(t, fromController(message), "valid controller message")
+}
+
+func TestGetApplicationFromValidControllerMessage(t *testing.T) {
+	message := new(Message)
+	err := json.Unmarshal([]byte(validControllerMessage), message)
+	assert.NoError(t, err, "error occured parsing log message")
+	expected := getApplicationFromControllerMessage(message)
+	assert.Equal(t, expected, "foo", "failed to retrieve app from message")
+}
+
+func TestBuildControllerLogMessageFromValidMessage(t *testing.T) {
+	message := new(Message)
+	err := json.Unmarshal([]byte(validControllerMessage), message)
+	assert.NoError(t, err, "error occured parsing log message")
+	expected := buildControllerLogMessage(message)
+	assert.Equal(t, expected,
+		"2016-10-18T20:29:38+00:00 deis[controller]: INFO admin deployed 2fd9226",
+		"failed to build controller log")
+}
+
+func TestBuildApplicationLogMessageFromValidMessage(t *testing.T) {
+	message := new(Message)
+	err := json.Unmarshal([]byte(validAppMessage), message)
+	assert.NoError(t, err, "error occured parsing log message")
+	expected := buildApplicationLogMessage(message)
+	assert.Equal(t, expected,
+		"2016-10-18T20:29:38+00:00 foo[web.v2.nzf60]: test message",
+		"failed to build application log")
+}
+
+func TestHandleValidAppMessage(t *testing.T) {
+	a, err := storage.NewRingBufferAdapter(1)
+	assert.NoError(t, err, "error creating ring buffer")
+	err = handle([]byte(validAppMessage), a)
+	assert.NoError(t, err, "error occured storing log message")
+	expected, _ := a.Read("foo", 1)
+	assert.Equal(t, expected[0],
+		"2016-10-18T20:29:38+00:00 foo[web.v2.nzf60]: test message",
+		"failed to aquire application log message")
+}
+
+func TestHandleValidControllerMessage(t *testing.T) {
+	a, err := storage.NewRingBufferAdapter(1)
+	assert.NoError(t, err, "error creating ring buffer")
+	err = handle([]byte(validControllerMessage), a)
+	assert.NoError(t, err, "error occured storing log message")
+	expected, _ := a.Read("foo", 1)
+	assert.Equal(t, expected[0],
+		"2016-10-18T20:29:38+00:00 deis[controller]: INFO admin deployed 2fd9226",
+		"failed to aquire controller log message")
+}
+
+func TestHandleInvalidAppMessage(t *testing.T) {
+	a, err := storage.NewRingBufferAdapter(1)
+	assert.NoError(t, err, "error creating ring buffer")
+	err = handle([]byte(validAppMessage), a)
+	assert.NoError(t, err, "error occured storing log message")
+	expected, _ := a.Read("foo", 1)
+	assert.Equal(t, expected[0],
+		"2016-10-18T20:29:38+00:00 foo[web.v2.nzf60]: test message",
+		"failed to aquire application log message")
+}
+
+func TestHandleInvalidControllerMessage(t *testing.T) {
+	a, err := storage.NewRingBufferAdapter(1)
+	assert.NoError(t, err, "error creating ring buffer")
+	err = handle([]byte(badjson), a)
+	assert.Error(t, err, "no error occured parsing json")
+}

log/model.go

@@ -1,5 +1,7 @@
 package log
 
+import "time"
+
 // Example log message JSON:
 //
 // {"log"=>"2016/05/31 01:34:43 10.164.1.1 GET / - 5074209722772702441\n", "stream"=>"stderr",
@@ -15,6 +17,7 @@ type Message struct {
 	Stream     string     `json:"stream"`
 	Kubernetes Kubernetes `json:"kubernetes"`
 	Docker     Docker     `json:"docker"`
+	Time       time.Time  `json:"time"`
 }
 
 // Kubernetes specific log message fields

manifests/deis-logger-deployment.yaml

@@ -0,0 +1,52 @@
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+ name: deis-logger
+ namespace: deis
+ labels:
+   heritage: deis
+ annotations:
+   helm-keep: "true"
+   component.deis.io/version: canary
+spec:
+  replicas: 1
+  strategy:
+    rollingUpdate:
+      maxSurge: 1
+      maxUnavailable: 0
+    type: RollingUpdate
+  selector:
+    matchLabels:
+      app: deis-logger
+  template:
+    metadata:
+      labels:
+        app: deis-logger
+    spec:
+      containers:
+      - name: deis-logger
+        image: quay.io/deis/logger:canary
+        imagePullPolicy: Always
+        env:
+        - name: STORAGE_ADAPTER
+          value: redis
+        - name: DEIS_LOGGER_REDIS_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: logger-redis-creds
+              key: password
+        ports:
+        - containerPort: 8088
+          name: http
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8088
+          initialDelaySeconds: 1
+          timeoutSeconds: 1
+        readinessProbe:
+          httpGet:
+            path: /healthz
+            port: 8088
+          initialDelaySeconds: 1
+          timeoutSeconds: 1