[server/kit] Adding automatic Stackdriver logging for App Engine 2nd …

…generation runtime (#159)

* attempting to use stackdriver combined access/app logging in GAE 2nd gen runtime

* adding new gae log file

* setting traceid

* letting sd client do the jsoning

* fixing silliness

* adding missed project, var args ...

* no need to make the logger public

* docs update

* adding log levels

* removing unused key

* closing the sd logger at shutdown

* removing zone

server/kit/README.md

@@ -3,9 +3,8 @@
 * The rationale behind this package:
     * A more opinionated server with fewer choices.
     * go-kit is used for serving HTTP/JSON & gRPC is used for serving HTTP2/RPC
-    * Monitoring and metrics are handled by a sidecar (ie. Cloud Endpoints)
-    * Logs always go to stdout/stderr
+    * Logs always go to stdout/stderr by default, but if running on App Engine, trace enabled Stackdriver logging will be used instead.
     * Using Go's 1.8 graceful HTTP shutdown
-    * Services using this package are meant for deploy to GCP with GKE and Cloud Endpoints.
+    * Services using this package are meant for deploy to GCP.
 
-* If you experience any issues please create an issue and/or reach out on the #gizmo channel with what you've found 
+* If you experience any issues please create an issue and/or reach out on the #gizmo channel with what you've found.

server/kit/gae_log.go

@@ -0,0 +1,167 @@
+package kit
+
+import (
+	"context"
+	"encoding"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+
+	"cloud.google.com/go/logging"
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/pkg/errors"
+	"google.golang.org/genproto/googleapis/api/monitoredres"
+)
+
+type gaeLogger struct {
+	project string
+	monRes  *monitoredres.MonitoredResource
+	lc      *logging.Client
+	lgr     *logging.Logger
+}
+
+func newAppEngineLogger(ctx context.Context, projectID, service, version string) (log.Logger, func() error, error) {
+	client, err := logging.NewClient(ctx, fmt.Sprintf("projects/%s", projectID))
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "unable to initiate stackdriver log client")
+	}
+	return gaeLogger{
+		lc:      client,
+		lgr:     client.Logger("app_logs"),
+		project: projectID,
+		monRes: &monitoredres.MonitoredResource{
+			Labels: map[string]string{
+				"module_id":  service,
+				"project_id": projectID,
+				"version_id": version,
+			},
+			Type: "gae_app",
+		},
+	}, client.Close, nil
+}
+
+func (l gaeLogger) Log(keyvals ...interface{}) error {
+	kvs, lvl, traceContext := logKeyValsToMap(keyvals...)
+	var traceID string
+	if traceContext != "" {
+		traceID = l.getTraceID(traceContext)
+	}
+
+	svrty := logging.Default
+	switch lvl {
+	case level.DebugValue():
+		svrty = logging.Debug
+	case level.ErrorValue():
+		svrty = logging.Error
+	case level.InfoValue():
+		svrty = logging.Info
+	case level.WarnValue():
+		svrty = logging.Warning
+	}
+
+	payload, err := json.Marshal(kvs)
+	if err != nil {
+		return err
+	}
+
+	l.lgr.Log(logging.Entry{
+		Severity: svrty,
+		Payload:  json.RawMessage(payload),
+		Trace:    traceID,
+		Resource: l.monRes,
+	})
+	return nil
+}
+
+func (l gaeLogger) getTraceID(traceCtx string) string {
+	return "projects/" + l.project + "/traces/" + strings.Split(traceCtx, "/")[0]
+}
+
+const cloudTraceLogKey = "cloud-trace"
+
+///////////////////////////////////////////////////
+// below funcs are straight up copied out of go-kit/kit/log:
+// https://github.com/go-kit/kit/blob/master/log/json_logger.go
+// we needed the magic for keyvals => map[string]interface{} but we're doing the
+// writing the JSON ourselves
+///////////////////////////////////////////////////
+
+func logKeyValsToMap(keyvals ...interface{}) (map[string]interface{}, level.Value, string) {
+	var (
+		lvl          level.Value
+		traceContext string
+	)
+	n := (len(keyvals) + 1) / 2 // +1 to handle case when len is odd
+	m := make(map[string]interface{}, n)
+	for i := 0; i < len(keyvals); i += 2 {
+		k := keyvals[i]
+		var v interface{} = log.ErrMissingValue
+		if i+1 < len(keyvals) {
+			v = keyvals[i+1]
+		}
+		merge(m, k, v)
+		if k == cloudTraceLogKey {
+			traceContext = v.(string)
+		}
+		if k == level.Key() {
+			lvl = v.(level.Value)
+		}
+	}
+	return m, lvl, traceContext
+}
+
+func merge(dst map[string]interface{}, k, v interface{}) {
+	var key string
+	switch x := k.(type) {
+	case string:
+		key = x
+	case fmt.Stringer:
+		key = safeString(x)
+	default:
+		key = fmt.Sprint(x)
+	}
+
+	// We want json.Marshaler and encoding.TextMarshaller to take priority over
+	// err.Error() and v.String(). But json.Marshall (called later) does that by
+	// default so we force a no-op if it's one of those 2 case.
+	switch x := v.(type) {
+	case json.Marshaler:
+	case encoding.TextMarshaler:
+	case error:
+		v = safeError(x)
+	case fmt.Stringer:
+		v = safeString(x)
+	}
+
+	dst[key] = v
+}
+
+func safeString(str fmt.Stringer) (s string) {
+	defer func() {
+		if panicVal := recover(); panicVal != nil {
+			if v := reflect.ValueOf(str); v.Kind() == reflect.Ptr && v.IsNil() {
+				s = "NULL"
+			} else {
+				panic(panicVal)
+			}
+		}
+	}()
+	s = str.String()
+	return
+}
+
+func safeError(err error) (s interface{}) {
+	defer func() {
+		if panicVal := recover(); panicVal != nil {
+			if v := reflect.ValueOf(err); v.Kind() == reflect.Ptr && v.IsNil() {
+				s = nil
+			} else {
+				panic(panicVal)
+			}
+		}
+	}()
+	s = err.Error()
+	return
+}

server/kit/kitserver.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	stdlog "log"
 	"net"
 	"net/http"
 	"net/http/pprof"
@@ -20,7 +21,8 @@ import (
 
 // Server encapsulates all logic for registering and running a gizmo kit server.
 type Server struct {
-	logger log.Logger
+	logger   log.Logger
+	logClose func() error
 
 	mux Router
 
@@ -42,6 +44,12 @@ const (
 	varsKey contextKey = iota
 	// key for logger
 	logKey
+
+	// ContextKeyCloudTraceContext is a context key for storing and retrieving the
+	// inbound 'x-cloud-trace-context' header. This server will automatically look for
+	// and inject the value into the request context. If in the App Engine environment
+	// this will be used to enable combined access and application logs.
+	ContextKeyCloudTraceContext
 )
 
 // NewServer will create a new kit server for the given Service.
@@ -62,11 +70,33 @@ func NewServer(svc Service) *Server {
 		r = opt(r)
 	}
 
+	// check if we're running on GAE via env variables
+	projectID := os.Getenv("GOOGLE_CLOUD_PROJECT")
+	serviceID := os.Getenv("GAE_SERVICE")
+	svcVersion := os.Getenv("GAE_VERSION")
+
+	var (
+		err      error
+		lg       log.Logger
+		logClose func() error
+	)
+	// use the version variable to determine if we're in the GAE environment
+	if svcVersion == "" {
+		lg = log.NewJSONLogger(log.NewSyncWriter(os.Stdout))
+	} else {
+		lg, logClose, err = newAppEngineLogger(context.Background(),
+			projectID, serviceID, svcVersion)
+		if err != nil {
+			stdlog.Fatalf("unable to start up app engine logger: %s", err)
+		}
+	}
+
 	s := &Server{
-		cfg:    cfg,
-		mux:    r,
-		exit:   make(chan chan error),
-		logger: log.NewJSONLogger(log.NewSyncWriter(os.Stdout)),
+		cfg:      cfg,
+		mux:      r,
+		exit:     make(chan chan error),
+		logger:   lg,
+		logClose: logClose,
 	}
 	s.svr = &http.Server{
 		Handler:        s,
@@ -88,9 +118,12 @@ func (s *Server) register(svc Service) {
 	s.svc = svc
 	opts := []httptransport.ServerOption{
 		// populate context with helpful keys
-		httptransport.ServerBefore(
-			httptransport.PopulateRequestContext,
-		),
+		httptransport.ServerBefore(func(ctx context.Context, r *http.Request) context.Context {
+			ctx = httptransport.PopulateRequestContext(ctx, r)
+			// add google trace header to use in tracing and logging
+			return context.WithValue(ctx, ContextKeyCloudTraceContext,
+				r.Header.Get("X-Cloud-Trace-Context"))
+		}),
 		// inject the server logger into every request context
 		httptransport.ServerBefore(func(ctx context.Context, _ *http.Request) context.Context {
 			return context.WithValue(ctx, logKey, AddLogKeyVals(ctx, s.logger))
@@ -209,6 +242,12 @@ func (s *Server) start() error {
 		// stop the listener with timeout
 		ctx, cancel := context.WithTimeout(context.Background(), s.cfg.ShutdownTimeout)
 		defer cancel()
+		defer func() {
+			// flush the logger after server shuts down
+			if s.logClose != nil {
+				s.logClose()
+			}
+		}()
 
 		if shutdown, ok := s.svc.(Shutdowner); ok {
 			shutdown.Shutdown()

server/kit/log.go

@@ -35,6 +35,7 @@ func AddLogKeyVals(ctx context.Context, l log.Logger) log.Logger {
 		http.ContextKeyRequestRemoteAddr:    "http-remote-addr",
 		http.ContextKeyRequestXForwardedFor: "http-x-forwarded-for",
 		http.ContextKeyRequestUserAgent:     "http-user-agent",
+		ContextKeyCloudTraceContext:         cloudTraceLogKey,
 	}
 	for k, v := range keys {
 		if val, ok := ctx.Value(k).(string); ok && val != "" {
@@ -53,13 +54,13 @@ func AddLogKeyVals(ctx context.Context, l log.Logger) log.Logger {
 }
 
 // LogMsg will log the given message to the server logger
-// with the key "msg" along with all the common request headers or gRPC metadata.
-func LogMsg(ctx context.Context, msg string) error {
-	return Logger(ctx).Log("msg", msg)
+// with the key "message" along with all the common request headers or gRPC metadata.
+func LogMsg(ctx context.Context, message string) error {
+	return Logger(ctx).Log("message", message)
 }
 
 // LogErrorMsg will log the given error under the key "error", the given message under
-// the key "msg" along with all the common request headers or gRPC metadata.
-func LogErrorMsg(ctx context.Context, err error, msg string) error {
-	return Logger(ctx).Log("error", err, "msg", msg)
+// the key "message" along with all the common request headers or gRPC metadata.
+func LogErrorMsg(ctx context.Context, err error, message string) error {
+	return Logger(ctx).Log("error", err, "message", message)
 }