✨ Add AccessLogInterceptor

Signed-off-by: Rintaro Okamura <[email protected]>

charts/vald/values.yaml

@@ -156,7 +156,7 @@ defaults:
             # @schema {"name": "defaults.server_config.servers.grpc.server.grpc.header_table_size", "type": "integer"}
             # defaults.server_config.servers.grpc.server.grpc.header_table_size -- gRPC server header table size
             header_table_size: 0
-            # @schema {"name": "defaults.server_config.servers.grpc.server.grpc.interceptors", "type": "array", "items": {"type": "string", "enum": ["RecoverInterceptor", "TracePayloadInterceptor"]}}
+            # @schema {"name": "defaults.server_config.servers.grpc.server.grpc.interceptors", "type": "array", "items": {"type": "string", "enum": ["RecoverInterceptor", "AccessLogInterceptor", "TracePayloadInterceptor"]}}
             # defaults.server_config.servers.grpc.server.grpc.interceptors -- gRPC server interceptors
             interceptors:
               - "RecoverInterceptor"

internal/net/grpc/interceptor/server/logging/accesslog.go

@@ -0,0 +1,152 @@
+//
+// Copyright (C) 2019-2021 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+// Package logging provides gRPC interceptors for access logging
+package logging
+
+import (
+	"context"
+	"path"
+	"time"
+
+	"github.com/vdaas/vald/internal/log"
+	"github.com/vdaas/vald/internal/net/grpc"
+	"github.com/vdaas/vald/internal/observability/trace"
+)
+
+const (
+	grpcKindUnary  = "unary"
+	grpcKindStream = "stream"
+
+	rpcCompletedMessage = "rpc completed"
+)
+
+type AccessLogEntity struct {
+	GRPC      *AccessLogGRPCEntity `json:"grpc,omitempty" yaml:"grpc"`
+	StartTime int64                `json:"startTime,omitempty" yaml:"startTime"`
+	Latency   int64                `json:"latency,omitempty" yaml:"latency"`
+	TraceID   string               `json:"traceID,omitempty" yaml:"traceID"`
+	Error     error                `json:"error,omitempty" yaml:"error"`
+}
+
+type AccessLogGRPCEntity struct {
+	Kind    string `json:"kind,omitempty" yaml:"kind"`
+	Service string `json:"service,omitempty" yaml:"service"`
+	Method  string `json:"method,omitempty" yaml:"method"`
+}
+
+func AccessLogInterceptor() grpc.UnaryServerInterceptor {
+	return func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (resp interface{}, err error) {
+		var traceID string
+
+		span := trace.FromContext(ctx)
+		if span != nil {
+			traceID = span.SpanContext().TraceID.String()
+		}
+
+		start := time.Now()
+
+		resp, err = handler(ctx, req)
+
+		latency := time.Since(start).Nanoseconds()
+		startTime := start.UnixNano()
+
+		service, method := parseMethod(info.FullMethod)
+
+		entity := &AccessLogEntity{
+			GRPC: &AccessLogGRPCEntity{
+				Kind:    grpcKindUnary,
+				Service: service,
+				Method:  method,
+			},
+			StartTime: startTime,
+			Latency:   latency,
+		}
+
+		if traceID != "" {
+			entity.TraceID = traceID
+		}
+
+		if err != nil {
+			entity.Error = err
+			log.Errord(rpcCompletedMessage, entity)
+		} else {
+			log.Infod(rpcCompletedMessage, entity)
+		}
+
+		return resp, err
+	}
+}
+
+func AccessLogStreamInterceptor() grpc.StreamServerInterceptor {
+	return func(
+		srv interface{},
+		ss grpc.ServerStream,
+		info *grpc.StreamServerInfo,
+		handler grpc.StreamHandler,
+	) error {
+		var traceID string
+
+		span := trace.FromContext(ss.Context())
+		if span != nil {
+			traceID = span.SpanContext().TraceID.String()
+		}
+
+		start := time.Now()
+
+		err := handler(srv, ss)
+
+		latency := time.Since(start).Nanoseconds()
+		startTime := start.UnixNano()
+
+		service, method := parseMethod(info.FullMethod)
+
+		entity := &AccessLogEntity{
+			GRPC: &AccessLogGRPCEntity{
+				Kind:    grpcKindStream,
+				Service: service,
+				Method:  method,
+			},
+			StartTime: startTime,
+			Latency:   latency,
+		}
+
+		if traceID != "" {
+			entity.TraceID = traceID
+		}
+
+		if err != nil {
+			entity.Error = err
+			log.Errord(rpcCompletedMessage, entity)
+		} else {
+			log.Infod(rpcCompletedMessage, entity)
+		}
+
+		return err
+	}
+}
+
+func parseMethod(fullMethod string) (service, method string) {
+	service = path.Dir(fullMethod)[1:]
+	method = path.Base(fullMethod)
+
+	return service, method
+}

internal/net/grpc/interceptor/server/logging/accesslog_test.go

@@ -0,0 +1,233 @@
+//
+// Copyright (C) 2019-2021 vdaas.org vald team <[email protected]>
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+// Package logging provides gRPC interceptors for access logging
+package logging
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/net/grpc"
+	"go.uber.org/goleak"
+)
+
+func TestAccessLogInterceptor(t *testing.T) {
+	t.Parallel()
+	type want struct {
+		want grpc.UnaryServerInterceptor
+	}
+	type test struct {
+		name       string
+		want       want
+		checkFunc  func(want, grpc.UnaryServerInterceptor) error
+		beforeFunc func()
+		afterFunc  func()
+	}
+	defaultCheckFunc := func(w want, got grpc.UnaryServerInterceptor) error {
+		if !reflect.DeepEqual(got, w.want) {
+			return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", got, w.want)
+		}
+		return nil
+	}
+	tests := []test{
+		// TODO test cases
+		/*
+		   {
+		       name: "test_case_1",
+		       want: want{},
+		       checkFunc: defaultCheckFunc,
+		   },
+		*/
+
+		// TODO test cases
+		/*
+		   func() test {
+		       return test {
+		           name: "test_case_2",
+		           want: want{},
+		           checkFunc: defaultCheckFunc,
+		       }
+		   }(),
+		*/
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc()
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc()
+			}
+			if test.checkFunc == nil {
+				test.checkFunc = defaultCheckFunc
+			}
+
+			got := AccessLogInterceptor()
+			if err := test.checkFunc(test.want, got); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+
+		})
+	}
+}
+
+func TestAccessLogStreamInterceptor(t *testing.T) {
+	t.Parallel()
+	type want struct {
+		want grpc.StreamServerInterceptor
+	}
+	type test struct {
+		name       string
+		want       want
+		checkFunc  func(want, grpc.StreamServerInterceptor) error
+		beforeFunc func()
+		afterFunc  func()
+	}
+	defaultCheckFunc := func(w want, got grpc.StreamServerInterceptor) error {
+		if !reflect.DeepEqual(got, w.want) {
+			return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", got, w.want)
+		}
+		return nil
+	}
+	tests := []test{
+		// TODO test cases
+		/*
+		   {
+		       name: "test_case_1",
+		       want: want{},
+		       checkFunc: defaultCheckFunc,
+		   },
+		*/
+
+		// TODO test cases
+		/*
+		   func() test {
+		       return test {
+		           name: "test_case_2",
+		           want: want{},
+		           checkFunc: defaultCheckFunc,
+		       }
+		   }(),
+		*/
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc()
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc()
+			}
+			if test.checkFunc == nil {
+				test.checkFunc = defaultCheckFunc
+			}
+
+			got := AccessLogStreamInterceptor()
+			if err := test.checkFunc(test.want, got); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+
+		})
+	}
+}
+
+func Test_parseMethod(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		fullMethod string
+	}
+	type want struct {
+		wantService string
+		wantMethod  string
+	}
+	type test struct {
+		name       string
+		args       args
+		want       want
+		checkFunc  func(want, string, string) error
+		beforeFunc func(args)
+		afterFunc  func(args)
+	}
+	defaultCheckFunc := func(w want, gotService string, gotMethod string) error {
+		if !reflect.DeepEqual(gotService, w.wantService) {
+			return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", gotService, w.wantService)
+		}
+		if !reflect.DeepEqual(gotMethod, w.wantMethod) {
+			return errors.Errorf("got: \"%#v\",\n\t\t\t\twant: \"%#v\"", gotMethod, w.wantMethod)
+		}
+		return nil
+	}
+	tests := []test{
+		// TODO test cases
+		/*
+		   {
+		       name: "test_case_1",
+		       args: args {
+		           fullMethod: "",
+		       },
+		       want: want{},
+		       checkFunc: defaultCheckFunc,
+		   },
+		*/
+
+		// TODO test cases
+		/*
+		   func() test {
+		       return test {
+		           name: "test_case_2",
+		           args: args {
+		           fullMethod: "",
+		           },
+		           want: want{},
+		           checkFunc: defaultCheckFunc,
+		       }
+		   }(),
+		*/
+	}
+
+	for _, tc := range tests {
+		test := tc
+		t.Run(test.name, func(tt *testing.T) {
+			tt.Parallel()
+			defer goleak.VerifyNone(tt, goleak.IgnoreCurrent())
+			if test.beforeFunc != nil {
+				test.beforeFunc(test.args)
+			}
+			if test.afterFunc != nil {
+				defer test.afterFunc(test.args)
+			}
+			if test.checkFunc == nil {
+				test.checkFunc = defaultCheckFunc
+			}
+
+			gotService, gotMethod := parseMethod(test.args.fullMethod)
+			if err := test.checkFunc(test.want, gotService, gotMethod); err != nil {
+				tt.Errorf("error = %v", err)
+			}
+
+		})
+	}
+}