Send request logs from activator.

cmd/activator/main.go

@@ -22,10 +22,9 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"os"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/knative/pkg/configmap"
 	"github.com/knative/pkg/controller"
 	"github.com/knative/pkg/logging/logkey"
@@ -47,6 +46,7 @@ import (
 	clientset "github.com/knative/serving/pkg/client/clientset/versioned"
 	servinginformers "github.com/knative/serving/pkg/client/informers/externalversions"
 	"github.com/knative/serving/pkg/goversion"
+	pkghttp "github.com/knative/serving/pkg/http"
 	"github.com/knative/serving/pkg/http/h2c"
 	"github.com/knative/serving/pkg/logging"
 	"github.com/knative/serving/pkg/metrics"
@@ -56,8 +56,8 @@ import (
 	"github.com/knative/serving/pkg/resources"
 	"github.com/knative/serving/pkg/tracing"
 	tracingconfig "github.com/knative/serving/pkg/tracing/config"
-
 	"github.com/openzipkin/zipkin-go"
+	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	kubeinformers "k8s.io/client-go/informers"
@@ -100,11 +100,13 @@ const (
 )
 
 var (
-	masterURL  = flag.String("master", "", "The address of the Kubernetes API server. Overrides any value in kubeconfig. Only required if out-of-cluster.")
+	masterURL = flag.String("master", "", "The address of the Kubernetes API server. "+
+		"Overrides any value in kubeconfig. Only required if out-of-cluster.")
 	kubeconfig = flag.String("kubeconfig", "", "Path to a kubeconfig. Only required if out-of-cluster.")
 )
 
-func statReporter(statSink *websocket.ManagedConnection, stopCh <-chan struct{}, statChan <-chan *autoscaler.StatMessage, logger *zap.SugaredLogger) {
+func statReporter(statSink *websocket.ManagedConnection, stopCh <-chan struct{},
+	statChan <-chan *autoscaler.StatMessage, logger *zap.SugaredLogger) {
 	for {
 		select {
 		case sm := <-statChan:
@@ -295,13 +297,21 @@ func main() {
 	ah = activatorhandler.NewRequestEventHandler(reqChan, ah)
 	ah = tracing.HTTPSpanMiddleware("handle_request", ah)
 	ah = configStore.HTTPMiddleware(ah)
+	reqLogHandler, err := pkghttp.NewRequestLogHandler(ah, logging.NewSyncFileWriter(os.Stdout), "",
+		requestLogTemplateInputGetter(revisionGetter))
+	if err != nil {
+		logger.Fatalw("Unable to create request log handler", zap.Error(err))
+	}
+	ah = reqLogHandler
 	ah = &activatorhandler.HealthHandler{HealthCheck: statSink.Status, NextHandler: ah}
 	ah = &activatorhandler.ProbeHandler{NextHandler: ah}
 
 	// Watch the logging config map and dynamically update logging levels.
 	configMapWatcher.Watch(logging.ConfigMapName(), logging.UpdateLevelFromConfigMap(logger, atomicLevel, component))
 	// Watch the observability config map and dynamically update metrics exporter.
 	configMapWatcher.Watch(metrics.ObservabilityConfigName, metrics.UpdateExporterFromConfigMap(component, logger))
+	// Watch the observability config map and dynamically update request logs.
+	configMapWatcher.Watch(metrics.ObservabilityConfigName, updateRequestLogFromConfigMap(logger, reqLogHandler))
 	if err = configMapWatcher.Start(stopCh); err != nil {
 		logger.Fatalw("Failed to start configuration manager", zap.Error(err))
 	}
@@ -327,5 +337,7 @@ func main() {
 
 func flush(logger *zap.SugaredLogger) {
 	logger.Sync()
+	os.Stdout.Sync()
+	os.Stderr.Sync()
 	pkgmetrics.FlushExporter()
 }

cmd/activator/request_log.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"net/http"
+
+	"github.com/knative/serving/pkg/activator"
+	"github.com/knative/serving/pkg/apis/serving"
+	pkghttp "github.com/knative/serving/pkg/http"
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func updateRequestLogFromConfigMap(logger *zap.SugaredLogger, h *pkghttp.RequestLogHandler) func(configMap *corev1.ConfigMap) {
+	return func(configMap *corev1.ConfigMap) {
+		newTemplate := configMap.Data["logging.request-log-template"]
+		if err := h.SetTemplate(newTemplate); err != nil {
+			logger.Errorw("Failed to update the request log template.", zap.Error(err), "template", newTemplate)
+		} else {
+			logger.Infow("Updated the request log template.", "template", newTemplate)
+		}
+	}
+}
+
+func requestLogTemplateInputGetter(getRevision activator.RevisionGetter) pkghttp.RequestLogTemplateInputGetter {
+	return func(req *http.Request, resp *pkghttp.RequestLogResponse) *pkghttp.RequestLogTemplateInput {
+		namespace := pkghttp.LastHeaderValue(req.Header, activator.RevisionHeaderNamespace)
+		name := pkghttp.LastHeaderValue(req.Header, activator.RevisionHeaderName)
+		revInfo := &pkghttp.RequestLogRevision{
+			Namespace: namespace,
+			Name:      name,
+		}
+
+		revision, err := getRevision(activator.RevisionID{Namespace: namespace, Name: name})
+		if err == nil && revision.Labels != nil {
+			revInfo.Configuration = revision.Labels[serving.ConfigurationLabelKey]
+			revInfo.Service = revision.Labels[serving.ServiceLabelKey]
+		}
+
+		return &pkghttp.RequestLogTemplateInput{
+			Request:  req,
+			Response: resp,
+			Revision: revInfo,
+		}
+	}
+}

cmd/activator/request_log_test.go

@@ -0,0 +1,180 @@
+package main
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	testing2 "github.com/knative/pkg/logging/testing"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/google/go-cmp/cmp"
+
+	"github.com/pkg/errors"
+
+	"github.com/knative/serving/pkg/activator"
+	"github.com/knative/serving/pkg/apis/serving"
+	"github.com/knative/serving/pkg/apis/serving/v1alpha1"
+	pkghttp "github.com/knative/serving/pkg/http"
+)
+
+const (
+	testRevisionName  = "testRevision"
+	testNamespaceName = "testNs"
+	testServiceName   = "testSvc"
+	testConfigName    = "testConfig"
+)
+
+func TestUpdateRequestLogFromConfigMap(t *testing.T) {
+	baseHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
+	buf := bytes.NewBufferString("")
+	handler, err := pkghttp.NewRequestLogHandler(baseHandler, buf, "",
+		requestLogTemplateInputGetter(getRevisionMock(true)))
+	if err != nil {
+		t.Fatalf("want: no error, got: %v", err)
+	}
+
+	tests := []struct {
+		name     string
+		url      string
+		body     string
+		template string
+		want     string
+	}{{
+		name:     "empty template",
+		url:      "http://example.com/testpage",
+		body:     "test",
+		template: "",
+		want:     "",
+	}, {
+		name:     "template with new line",
+		url:      "http://example.com/testpage",
+		body:     "test",
+		template: "{{.Request.URL}}\n",
+		want:     "http://example.com/testpage\n",
+	}, {
+		name:     "invalid template",
+		url:      "http://example.com",
+		body:     "test",
+		template: "{{}}",
+		want:     "http://example.com\n",
+	}, {
+		name:     "revision info",
+		url:      "http://example.com",
+		body:     "test",
+		template: "{{.Revision.Name}}, {{.Revision.Namespace}}, {{.Revision.Service}}, {{.Revision.Configuration}}, {{.Revision.PodName}}, {{.Revision.PodIP}}",
+		want:     "testRevision, testNs, testSvc, testConfig, , \n",
+	}, {
+		name:     "empty template 2",
+		url:      "http://example.com/testpage",
+		body:     "test",
+		template: "",
+		want:     "",
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			buf.Reset()
+			cm := &corev1.ConfigMap{}
+			cm.Data = map[string]string{"logging.request-log-template": test.template}
+			(updateRequestLogFromConfigMap(testing2.TestLogger(t), handler))(cm)
+			resp := httptest.NewRecorder()
+			req := httptest.NewRequest(http.MethodPost, test.url, bytes.NewBufferString(test.body))
+			req.Header = map[string][]string{
+				http.CanonicalHeaderKey(activator.RevisionHeaderName):      {testRevisionName},
+				http.CanonicalHeaderKey(activator.RevisionHeaderNamespace): {testNamespaceName},
+			}
+			handler.ServeHTTP(resp, req)
+
+			got := string(buf.Bytes())
+			if got != test.want {
+				t.Errorf("got '%v', want '%v'", got, test.want)
+			}
+		})
+	}
+}
+
+func TestRequestLogTemplateInputGetter(t *testing.T) {
+	tests := []struct {
+		name     string
+		getter   pkghttp.RequestLogTemplateInputGetter
+		request  *http.Request
+		response *pkghttp.RequestLogResponse
+		want     pkghttp.RequestLogRevision
+	}{{
+		name:   "success",
+		getter: requestLogTemplateInputGetter(getRevisionMock(true)),
+		request: &http.Request{Header: map[string][]string{
+			http.CanonicalHeaderKey(activator.RevisionHeaderName):      {testRevisionName},
+			http.CanonicalHeaderKey(activator.RevisionHeaderNamespace): {testNamespaceName},
+		}},
+		response: &pkghttp.RequestLogResponse{Code: http.StatusAlreadyReported},
+		want: pkghttp.RequestLogRevision{
+			Namespace:     testNamespaceName,
+			Name:          testRevisionName,
+			Configuration: testConfigName,
+			Service:       testServiceName,
+		},
+	}, {
+		name:   "revision not found",
+		getter: requestLogTemplateInputGetter(getRevisionMock(true)),
+		request: &http.Request{Header: map[string][]string{
+			http.CanonicalHeaderKey(activator.RevisionHeaderName):      {"foo"},
+			http.CanonicalHeaderKey(activator.RevisionHeaderNamespace): {"bar"},
+		}},
+		response: &pkghttp.RequestLogResponse{Code: http.StatusAlreadyReported},
+		want: pkghttp.RequestLogRevision{
+			Name:      "foo",
+			Namespace: "bar",
+		},
+	}, {
+		name:   "labels not found",
+		getter: requestLogTemplateInputGetter(getRevisionMock(false)),
+		request: &http.Request{Header: map[string][]string{
+			http.CanonicalHeaderKey(activator.RevisionHeaderName):      {testRevisionName},
+			http.CanonicalHeaderKey(activator.RevisionHeaderNamespace): {testNamespaceName},
+		}},
+		response: &pkghttp.RequestLogResponse{Code: http.StatusAlreadyReported},
+		want: pkghttp.RequestLogRevision{
+			Namespace: testNamespaceName,
+			Name:      testRevisionName,
+		},
+	}}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.getter(test.request, test.response)
+			if !cmp.Equal(*got.Revision, test.want) {
+				t.Errorf("Got = %v, want: %v, diff: %s", got.Revision, test.want, cmp.Diff(got.Revision, test.want))
+			}
+			if got.Request != test.request {
+				t.Errorf("Got = %v, want: %v", got.Request, test.request)
+			}
+			if got.Response != test.response {
+				t.Errorf("Got = %v, want: %v", got.Response, test.response)
+			}
+		})
+	}
+}
+
+func getRevisionMock(addLabels bool) func(revID activator.RevisionID) (*v1alpha1.Revision, error) {
+	return func(revID activator.RevisionID) (*v1alpha1.Revision, error) {
+		rev := &v1alpha1.Revision{}
+		if revID.Name != testRevisionName || revID.Namespace != testNamespaceName {
+			return nil, errors.New("revision not found")
+		}
+		rev.Name = revID.Name
+		rev.Namespace = revID.Namespace
+		if addLabels {
+			rev.Labels = map[string]string{
+				serving.ConfigurationLabelKey: testConfigName,
+				serving.ServiceLabelKey:       testServiceName,
+			}
+		}
+		return rev, nil
+	}
+}

cmd/queue/main.go

@@ -26,8 +26,6 @@ import (
 	"os"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/knative/pkg/logging/logkey"
 	"github.com/knative/pkg/metrics"
 	"github.com/knative/pkg/signals"
@@ -36,13 +34,14 @@ import (
 	activatorutil "github.com/knative/serving/pkg/activator/util"
 	"github.com/knative/serving/pkg/apis/networking"
 	"github.com/knative/serving/pkg/autoscaler"
+	pkghttp "github.com/knative/serving/pkg/http"
 	"github.com/knative/serving/pkg/http/h2c"
 	"github.com/knative/serving/pkg/logging"
 	"github.com/knative/serving/pkg/network"
 	"github.com/knative/serving/pkg/queue"
 	"github.com/knative/serving/pkg/queue/health"
 	queuestats "github.com/knative/serving/pkg/queue/stats"
-
+	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
@@ -86,7 +85,6 @@ var (
 	h2cProxy  *httputil.ReverseProxy
 	httpProxy *httputil.ReverseProxy
 
-	server           *http.Server
 	healthState      = &health.State{}
 	promStatReporter *queue.PrometheusStatsReporter // Prometheus stats reporter.
 )
@@ -329,15 +327,16 @@ func pushRequestLogHandler(currentHandler http.Handler) http.Handler {
 		return currentHandler
 	}
 
-	revInfo := &queue.RequestLogRevInfo{
+	revInfo := &pkghttp.RequestLogRevision{
 		Name:          servingRevision,
 		Namespace:     servingNamespace,
 		Service:       servingService,
 		Configuration: servingConfig,
 		PodName:       servingPodName,
 		PodIP:         servingPodIP,
 	}
-	handler, err := queue.NewRequestLogHandler(currentHandler, logging.NewSyncFileWriter(os.Stdout), templ, revInfo)
+	handler, err := pkghttp.NewRequestLogHandler(currentHandler, logging.NewSyncFileWriter(os.Stdout), templ,
+		pkghttp.RequestLogTemplateInputGetterFromRevision(revInfo))
 
 	if err != nil {
 		logger.Errorw("Error setting up request logger. Request logs will be unavailable.", zap.Error(err))