chore(transport): Refactor to use pluggable Aggregator component...

Author: krancour

Makefile

@@ -5,9 +5,8 @@ GOLINT = golint
 GOTEST = $(GO) test --cover --race -v
 GOVET = $(GO) vet
 GO_FILES = $(wildcard *.go)
-GO_PACKAGES = storage syslogish weblog
+GO_PACKAGES = storage log weblog
 GO_PACKAGES_REPO_PATH = $(addprefix $(REPO_PATH)/,$(GO_PACKAGES))
-GO_TESTABLE_PACKAGES_REPO_PATH = $(addprefix $(REPO_PATH)/,storage storage/file storage/ringbuffer)
 
 # the filepath to this repository, relative to $GOPATH/src
 REPO_PATH = github.com/deis/logger
@@ -50,21 +49,21 @@ bootstrap: check-docker
 build-binary:
 	GOOS=linux GOARCH=amd64 go build -ldflags ${LDFLAGS} -o $(BINARY_DEST_DIR)/logger .
 
-build: build-with-container docker-build
+build: docker-build
+build-without-container: build-binary build-image
+push: docker-push
+upgrade: kube-update
+install: kube-install
+uninstall: kube-delete
 
 # Containerized build of the binary
 build-with-container: check-docker
 	mkdir -p ${BINARY_DEST_DIR}
 	${DEV_ENV_CMD} make build-binary
-	docker build --rm -t ${IMAGE} rootfs
 
-build-without-container: build-binary
-	docker build -t ${IMAGE} rootfs
-	docker tag ${IMAGE} ${MUTABLE_IMAGE}
-
-push: docker-push
+docker-build: build-with-container build-image
 
-docker-build: build-with-container
+build-image:
 	docker build -t ${IMAGE} rootfs
 	docker tag ${IMAGE} ${MUTABLE_IMAGE}
 
@@ -91,22 +90,19 @@ style-check:
 	shellcheck $(SHELL_SCRIPTS)
 
 test-unit:
-	${DEV_ENV_CMD} $(GOTEST) $(GO_TESTABLE_PACKAGES_REPO_PATH)
+	${DEV_ENV_CMD} $(GOTEST) $$(glide nv)
 
 kube-install:
 	kubectl create -f manifests/deis-logger-svc.yaml
 	kubectl create -f manifests/deis-logger-rc.yaml
-	kubectl create -f manifests/deis-logger-fluentd-daemon.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-fluentd-daemon.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-fluentd-daemon.yaml
 
 kube-replace: build push update-manifests
 	kubectl replace --force -f manifests/deis-logger-rc.tmp.yaml

config.go

@@ -9,8 +9,9 @@ const (
 )
 
 type config struct {
-	StorageType string `envconfig:"STORAGE_ADAPTER" default:"memory"`
-	NumLines    int    `envconfig:"NUMBER_OF_LINES" default:"1000"`
+	StorageType    string `envconfig:"STORAGE_ADAPTER" default:"memory"`
+	NumLines       int    `envconfig:"NUMBER_OF_LINES" default:"1000"`
+	AggregatorType string `envconfig:"AGGREGATOR_TYPE" default:"nsq"`
 }
 
 func parseConfig(appName string) (*config, error) {

glide.lock

@@ -3,3 +3,4 @@ import:
 - package: github.com/gorilla/mux
 - package: github.com/gorilla/context
 - package: github.com/kelseyhightower/envconfig
+- package: github.com/nsqio/go-nsq

glide.yaml

@@ -0,0 +1,17 @@
+package log
+
+// Aggregator is an interface for pluggable components that collect log messages delivered via
+// some transport mechanism
+type Aggregator interface {
+	// Listen causes an aggregator to begin consuming messages via its underlying transport
+	// mechanism. Implementations of this must be non-blocking. Stop() can be called to stop the
+	// aggregator.
+	Listen() error
+	// Stop stops the consumer and blocks until it stops or the configured duration passes. In the
+	// latter case, returns an error of type ErrStopTimedOut
+	Stop() error
+	// Stopped returns a channel that will receive when the consumer has stopped. The error received
+	// may be nil, which means it was stopped cleanly. A non-nil error means it stopped because it
+	// errored.
+	Stopped() <-chan error
+}

log/aggregator.go

@@ -0,0 +1,16 @@
+package log
+
+import (
+	"fmt"
+
+	"github.com/deis/logger/storage"
+)
+
+// NewAggregator returns a pointer to an appropriate implementation of the Aggregator interface, as
+// determined by the aggregatorType string it is passed.
+func NewAggregator(aggregatorType string, storageAdapter storage.Adapter) (Aggregator, error) {
+	if aggregatorType == "nsq" {
+		return newNSQAggregator(storageAdapter), nil
+	}
+	return nil, fmt.Errorf("Unrecognized aggregator type: '%s'", aggregatorType)
+}

log/aggregator_factory.go

@@ -0,0 +1,45 @@
+package log
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+)
+
+type stubStorageAdapter struct {
+}
+
+func (a *stubStorageAdapter) Write(app string, message string) error {
+	return nil
+}
+
+func (a *stubStorageAdapter) Read(app string, lines int) ([]string, error) {
+	return []string{}, nil
+}
+
+func (a *stubStorageAdapter) Destroy(app string) error {
+	return nil
+}
+
+func (a *stubStorageAdapter) Reopen() error {
+	return nil
+}
+
+func TestGetUsingInvalidValues(t *testing.T) {
+	_, err := NewAggregator("bogus", &stubStorageAdapter{})
+	if err == nil || err.Error() != fmt.Sprintf("Unrecognized aggregator type: '%s'", "bogus") {
+		t.Error("Did not receive expected error message")
+	}
+}
+
+func TestNSQBasedAggregator(t *testing.T) {
+	a, err := NewAggregator("nsq", &stubStorageAdapter{})
+	if err != nil {
+		t.Error(err)
+	}
+	expected := "*log.nsqAggregator"
+	aType := reflect.TypeOf(a).String()
+	if aType != expected {
+		t.Errorf("Expected a %s, but got a %s", expected, aType)
+	}
+}

log/aggregator_factory_test.go

@@ -0,0 +1,37 @@
+package log
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/kelseyhightower/envconfig"
+)
+
+const (
+	appName = "logger"
+)
+
+type config struct {
+	NSQHost            string `envconfig:"DEIS_NSQD_SERVICE_HOST" default:""`
+	NSQPort            int    `envconfig:"DEIS_NSQD_SERVICE_PORT_TRANSPORT" default:"4150"`
+	NSQTopic           string `envconfig:"NSQ_TOPIC" default:"logs"`
+	NSQChannel         string `envconfig:"NSQ_CHANNEL" default:"consume"`
+	NSQHandlerCount    int    `envconfig:"NSQ_HANDLER_COUNT" default:"30"`
+	StopTimeoutSeconds int    `envconfig:"AGGREGATOR_STOP_TIMEOUT_SEC" default:"1"`
+}
+
+func (c config) nsqURL() string {
+	return fmt.Sprintf("%s:%d", c.NSQHost, c.NSQPort)
+}
+
+func (c config) stopTimeoutDuration() time.Duration {
+	return time.Duration(c.StopTimeoutSeconds) * time.Second
+}
+
+func parseConfig(appName string) (*config, error) {
+	ret := new(config)
+	if err := envconfig.Process(appName, ret); err != nil {
+		return nil, err
+	}
+	return ret, nil
+}

log/config.go

@@ -0,0 +1,20 @@
+package log
+
+import (
+	"fmt"
+	"time"
+)
+
+// ErrStopTimedOut is the error returned if a (Aggregator).Stop call times out before the stop is
+// complete
+type ErrStopTimedOut struct {
+	Timeout time.Duration
+}
+
+func newErrStopTimedOut(timeout time.Duration) ErrStopTimedOut {
+	return ErrStopTimedOut{Timeout: timeout}
+}
+
+func (e ErrStopTimedOut) Error() string {
+	return fmt.Sprintf("stopping a consumer timed out after %s", e.Timeout)
+}

log/errors.go

@@ -0,0 +1,47 @@
+package log
+
+import (
+	"encoding/json"
+	"fmt"
+	"regexp"
+
+	"github.com/deis/logger/storage"
+)
+
+const (
+	controllerPattern       = `^(INFO|WARN|DEBUG|ERROR)\s+(\[(\S+)\])+:(.*)`
+	controllerContainerName = "deis-controller"
+)
+
+var (
+	regex = regexp.MustCompile(controllerPattern)
+)
+
+func handle(rawMessage []byte, storageAdapter storage.Adapter) error {
+	message := new(Message)
+	if err := json.Unmarshal(rawMessage, message); err != nil {
+		return err
+	}
+	if fromController(message) {
+		storageAdapter.Write(getApplicationFromControllerMessage(message), message.Log)
+	} else {
+		labels := message.Kubernetes.Labels
+		storageAdapter.Write(labels["app"], buildApplicationLogMessage(message))
+	}
+	return nil
+}
+
+func fromController(message *Message) bool {
+	matched, _ := regexp.MatchString(controllerContainerName, message.Kubernetes.ContainerName)
+	return matched
+}
+
+func getApplicationFromControllerMessage(message *Message) string {
+	return regex.FindStringSubmatch(message.Log)[3]
+}
+
+func buildApplicationLogMessage(message *Message) string {
+	body := message.Log
+	podName := message.Kubernetes.PodName
+	return fmt.Sprintf("%s -- %s", podName, body)
+}