Skip to content

Commit

Permalink
Send request logs from activator. (#3927)
Browse files Browse the repository at this point in the history
* Send request logs from activator.

* Fix linter issues.
  • Loading branch information
mdemirhan authored and knative-prow-robot committed Apr 30, 2019
1 parent 6f3db8a commit 3bfdea5
Show file tree
Hide file tree
Showing 7 changed files with 524 additions and 129 deletions.
24 changes: 18 additions & 6 deletions cmd/activator/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -46,6 +45,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/logging"
"github.com/knative/serving/pkg/metrics"
"github.com/knative/serving/pkg/network"
Expand All @@ -54,8 +54,8 @@ import (
"github.com/knative/serving/pkg/resources"
"github.com/knative/serving/pkg/tracing"
tracingconfig "github.com/knative/serving/pkg/tracing/config"

zipkin "github.com/openzipkin/zipkin-go"
"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"
Expand Down Expand Up @@ -98,11 +98,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:
Expand Down Expand Up @@ -293,13 +295,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))
}
Expand All @@ -325,5 +335,7 @@ func main() {

func flush(logger *zap.SugaredLogger) {
logger.Sync()
os.Stdout.Sync()
os.Stderr.Sync()
pkgmetrics.FlushExporter()
}
61 changes: 61 additions & 0 deletions cmd/activator/request_log.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
/*
Copyright 2019 The Knative Authors
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
http://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 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,
}
}
}
196 changes: 196 additions & 0 deletions cmd/activator/request_log_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,196 @@
/*
Copyright 2019 The Knative Authors
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
http://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 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
}
}
11 changes: 5 additions & 6 deletions cmd/queue/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -36,12 +34,13 @@ 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/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"
)

Expand Down Expand Up @@ -85,7 +84,6 @@ var (
h2cProxy *httputil.ReverseProxy
httpProxy *httputil.ReverseProxy

server *http.Server
healthState = &health.State{}
promStatReporter *queue.PrometheusStatsReporter // Prometheus stats reporter.
)
Expand Down Expand Up @@ -328,15 +326,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))
Expand Down
Loading

0 comments on commit 3bfdea5

Please sign in to comment.