Merge pull request #748 from HotelsDotCom/feature/journald-feedback

Provide real-time feedback from Journald logs when updating/creating …

Author: mumoshu

Documentation/kubernetes-on-aws-journald-cloudwatch-logs.md

@@ -14,11 +14,43 @@ cloudWatchLogging:
  retentionInDays: 7
 ```
 
+
 The docker image is also configurable:
 
 ```
 journaldCloudWatchLogsImage:
   repo: "jollinshead/journald-cloudwatch-logs"
   tag: "0.1"
   rktPullDocker: true
-```
+```
+
+## kube-aws up/update feedback
+
+During kube-aws up/update, filtered Journald logs can be printed to stdout.
+This feature is configurable in cluster.yaml under the *cloudWatchLogging* section, and requires *cloudWatchLogging* to be enabled.
+( Default values: )
+
+```
+cloudWatchLogging:
+ enabled: false
+ imageWithTag: jollinshead/journald-cloudwatch-logs:0.1
+ retentionInDays: 7
+ localStreaming:
+  enabled: true,
+  filter:  `{ $.priority = "CRIT" || $.priority = "WARNING" && $.transport = "journal" && $.systemdUnit = "init.scope" }`,
+  interval: 60
+```
+
+NOTE: Due to high initial entropy, *.service* failures may occur during the early stages of booting.
+In this context Entropy refers to the disorder of *.service*s (starting, failing, restarting).
+
+### Parameters
+
+#### Filter
+By default the filter is configured for *.service* failures and messages flagged as 'critical'.
+See [the official AWS documentation](http://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/FilterAndPatternSyntax.html) for more information.
+
+#### Interval
+Since some messages are produced frequently, to avoid excessive spam, an 'interval' parameter is provided.
+This 'interval' value determines the time between printing two identical messages to stdout.
+

README.md

@@ -48,6 +48,7 @@ Check out our getting started tutorial on launching your first Kubernetes cluste
   * [Backup Kubernetes resources](/Documentation/kubernetes-on-aws-backup.md)
   * [Restore Kubernetes resources](/contrib/cluster-backup/README.md)
   * [Journald logging to AWS CloudWatch](/Documentation/kubernetes-on-aws-journald-cloudwatch-logs.md)
+    * [kube-aws up/update feedback](/Documentation/kubernetes-on-aws-journald-cloudwatch-logs.md)
 
 ## Examples
 

core/controlplane/config/config.go

@@ -113,6 +113,11 @@ func NewDefaultCluster() *Cluster {
 			CloudWatchLogging: CloudWatchLogging{
 				Enabled:         false,
 				RetentionInDays: 7,
+				LocalStreaming: LocalStreaming{
+					Enabled:  true,
+					Filter:   `{ $.priority = "CRIT" || $.priority = "WARNING" && $.transport = "journal" && $.systemdUnit = "init.scope" }`,
+					interval: 60,
+				},
 			},
 			HyperkubeImage:                     model.Image{Repo: "quay.io/coreos/hyperkube", Tag: k8sVer, RktPullDocker: false},
 			AWSCliImage:                        model.Image{Repo: "quay.io/coreos/awscli", Tag: "master", RktPullDocker: false},
@@ -747,6 +752,18 @@ type KubeResourcesAutosave struct {
 type CloudWatchLogging struct {
 	Enabled         bool `yaml:"enabled"`
 	RetentionInDays int  `yaml:"retentionInDays"`
+	LocalStreaming  `yaml:"localStreaming"`
+}
+
+type LocalStreaming struct {
+	Enabled  bool   `yaml:"enabled"`
+	Filter   string `yaml:"filter"`
+	interval int    `yaml:"interval"`
+}
+
+func (c *LocalStreaming) Interval() int64 {
+	// Convert from seconds to milliseconds (and return as int64 type)
+	return int64(c.interval * 1000)
 }
 
 func (c *CloudWatchLogging) MergeIfEmpty(other CloudWatchLogging) {

core/controlplane/config/templates/cluster.yaml

@@ -1145,6 +1145,12 @@ worker:
 #cloudWatchLogging:
 # enabled: false
 # retentionInDays: 7
+# # When enabled, feedback from Journald logs (with an applied filter) will be outputted during kube-aws 'apply | up'.
+# # It is enabled by default - provided cloudWatchLogging is enabled.
+# localStreaming:
+#  enabled: true,
+#  filter:  `{ $.priority = "CRIT" || $.priority = "WARNING" && $.transport = "journal" && $.systemdUnit = "init.scope" }`,
+#  interval: 60
 
 # Addon features
 addons:

core/root/cluster.go

@@ -1,10 +1,12 @@
 package root
 
 import (
+	"encoding/json"
 	"fmt"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/cloudformation"
+	"github.com/aws/aws-sdk-go/service/cloudwatchlogs"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/kubernetes-incubator/kube-aws/cfnstack"
 	controlplane "github.com/kubernetes-incubator/kube-aws/core/controlplane/cluster"
@@ -14,10 +16,12 @@ import (
 	"github.com/kubernetes-incubator/kube-aws/core/root/config"
 	"github.com/kubernetes-incubator/kube-aws/core/root/defaults"
 	"github.com/kubernetes-incubator/kube-aws/filereader/jsontemplate"
+	model "github.com/kubernetes-incubator/kube-aws/model"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 )
 
 const (
@@ -182,6 +186,13 @@ func (c clusterImpl) Create() error {
 		return err
 	}
 
+	if c.controlPlane.CloudWatchLogging.Enabled && c.controlPlane.CloudWatchLogging.LocalStreaming.Enabled {
+		// Return Journald logs in a separate GoRoutine
+		quit := make(chan bool)
+		defer func() { quit <- true }()
+		go streamJournaldLogs(c, quit)
+	}
+
 	return c.stackProvisioner().CreateStackAtURLAndWait(cfSvc, stackTemplateURL)
 }
 
@@ -296,6 +307,13 @@ func (c clusterImpl) Update() (string, error) {
 		return "", err
 	}
 
+	if c.controlPlane.CloudWatchLogging.Enabled && c.controlPlane.CloudWatchLogging.LocalStreaming.Enabled {
+		// Return Journald logs in a separate GoRoutine
+		quit := make(chan bool)
+		defer func() { quit <- true }()
+		go streamJournaldLogs(c, quit)
+	}
+
 	return c.stackProvisioner().UpdateStackAtURLAndWait(cfSvc, templateUrl)
 }
 
@@ -348,3 +366,42 @@ func (c clusterImpl) ValidateStack() (string, error) {
 
 	return strings.Join(reports, "\n"), nil
 }
+
+func streamJournaldLogs(c clusterImpl, quit chan bool) error {
+	fmt.Printf("Printing filtered Journald logs for log group '%s'...\nNOTE: Due to high initial entropy, failures may occur during the early stages of booting.\n", c.controlPlane.ClusterName)
+	cwlSvc := cloudwatchlogs.New(c.session)
+	startTime := time.Now().Unix() * 1E3
+	fleInput := cloudwatchlogs.FilterLogEventsInput{
+		LogGroupName:  &c.controlPlane.ClusterName,
+		FilterPattern: &c.controlPlane.CloudWatchLogging.LocalStreaming.Filter,
+		StartTime:     &startTime}
+	messages := make(map[string]int64)
+
+	for {
+		select {
+		case <-quit:
+			return nil
+		default:
+			out, err := cwlSvc.FilterLogEvents(&fleInput)
+			if err != nil {
+				continue
+			}
+			if len(out.Events) > 1 {
+				startTime = *out.Events[len(out.Events)-1].Timestamp
+				for _, event := range out.Events {
+					if *event.Timestamp > messages[*event.Message]+c.controlPlane.CloudWatchLogging.LocalStreaming.Interval() {
+						messages[*event.Message] = *event.Timestamp
+						res := model.SystemdMessageResponse{}
+						json.Unmarshal([]byte(*event.Message), &res)
+						fmt.Printf("%s: \"%s\"\n", res.Hostname, res.Message)
+					}
+				}
+			}
+			fleInput = cloudwatchlogs.FilterLogEventsInput{
+				LogGroupName:  &c.controlPlane.ClusterName,
+				FilterPattern: &c.controlPlane.CloudWatchLogging.LocalStreaming.Filter,
+				NextToken:     out.NextToken,
+				StartTime:     &startTime}
+		}
+	}
+}

model/cloudwatch_logging.go

@@ -0,0 +1,12 @@
+package model
+
+type SystemdMessageResponse struct {
+	InstanceId  string `json:"instanceId,omitempty"`
+	Hostname    string `json:"hostname,omitempty"`
+	CmdName     string `json:"cmdName,omitempty"`
+	Exe         string `json:"exe,omitempty"`
+	CmdLine     string `json:"cmdLine,omitempty"`
+	SystemdUnit string `json:"systemdUnit,omitempty"`
+	Priority    string `json:"priority,omitempty"`
+	Message     string `json:"message,omitempty"`
+}