218 lines
5.9 KiB
Go
218 lines
5.9 KiB
Go
// Copyright 2018, OpenCensus 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 ochttp
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"net/http"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
"go.opencensus.io/stats"
|
|
"go.opencensus.io/tag"
|
|
"go.opencensus.io/trace"
|
|
"go.opencensus.io/trace/propagation"
|
|
)
|
|
|
|
// Handler is a http.Handler that is aware of the incoming request's span.
|
|
//
|
|
// The extracted span can be accessed from the incoming request's
|
|
// context.
|
|
//
|
|
// span := trace.FromContext(r.Context())
|
|
//
|
|
// The server span will be automatically ended at the end of ServeHTTP.
|
|
//
|
|
// Incoming propagation mechanism is determined by the given HTTP propagators.
|
|
type Handler struct {
|
|
// Propagation defines how traces are propagated. If unspecified,
|
|
// B3 propagation will be used.
|
|
Propagation propagation.HTTPFormat
|
|
|
|
// Handler is the handler used to handle the incoming request.
|
|
Handler http.Handler
|
|
|
|
// StartOptions are applied to the span started by this Handler around each
|
|
// request.
|
|
//
|
|
// StartOptions.SpanKind will always be set to trace.SpanKindServer
|
|
// for spans started by this transport.
|
|
StartOptions trace.StartOptions
|
|
|
|
// IsPublicEndpoint should be set to true for publicly accessible HTTP(S)
|
|
// servers. If true, any trace metadata set on the incoming request will
|
|
// be added as a linked trace instead of being added as a parent of the
|
|
// current trace.
|
|
IsPublicEndpoint bool
|
|
}
|
|
|
|
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
var traceEnd, statsEnd func()
|
|
r, traceEnd = h.startTrace(w, r)
|
|
defer traceEnd()
|
|
w, statsEnd = h.startStats(w, r)
|
|
defer statsEnd()
|
|
handler := h.Handler
|
|
if handler == nil {
|
|
handler = http.DefaultServeMux
|
|
}
|
|
handler.ServeHTTP(w, r)
|
|
}
|
|
|
|
func (h *Handler) startTrace(w http.ResponseWriter, r *http.Request) (*http.Request, func()) {
|
|
name := spanNameFromURL(r.URL)
|
|
ctx := r.Context()
|
|
var span *trace.Span
|
|
sc, ok := h.extractSpanContext(r)
|
|
if ok && !h.IsPublicEndpoint {
|
|
ctx, span = trace.StartSpanWithRemoteParent(ctx, name, sc,
|
|
trace.WithSampler(h.StartOptions.Sampler),
|
|
trace.WithSpanKind(trace.SpanKindServer))
|
|
} else {
|
|
ctx, span = trace.StartSpan(ctx, name,
|
|
trace.WithSampler(h.StartOptions.Sampler),
|
|
trace.WithSpanKind(trace.SpanKindServer),
|
|
)
|
|
if ok {
|
|
span.AddLink(trace.Link{
|
|
TraceID: sc.TraceID,
|
|
SpanID: sc.SpanID,
|
|
Type: trace.LinkTypeChild,
|
|
Attributes: nil,
|
|
})
|
|
}
|
|
}
|
|
span.AddAttributes(requestAttrs(r)...)
|
|
return r.WithContext(ctx), span.End
|
|
}
|
|
|
|
func (h *Handler) extractSpanContext(r *http.Request) (trace.SpanContext, bool) {
|
|
if h.Propagation == nil {
|
|
return defaultFormat.SpanContextFromRequest(r)
|
|
}
|
|
return h.Propagation.SpanContextFromRequest(r)
|
|
}
|
|
|
|
func (h *Handler) startStats(w http.ResponseWriter, r *http.Request) (http.ResponseWriter, func()) {
|
|
ctx, _ := tag.New(r.Context(),
|
|
tag.Upsert(Host, r.URL.Host),
|
|
tag.Upsert(Path, r.URL.Path),
|
|
tag.Upsert(Method, r.Method))
|
|
track := &trackingResponseWriter{
|
|
start: time.Now(),
|
|
ctx: ctx,
|
|
writer: w,
|
|
}
|
|
if r.Body == nil {
|
|
// TODO: Handle cases where ContentLength is not set.
|
|
track.reqSize = -1
|
|
} else if r.ContentLength > 0 {
|
|
track.reqSize = r.ContentLength
|
|
}
|
|
stats.Record(ctx, ServerRequestCount.M(1))
|
|
return track, track.end
|
|
}
|
|
|
|
type trackingResponseWriter struct {
|
|
ctx context.Context
|
|
reqSize int64
|
|
respSize int64
|
|
start time.Time
|
|
statusCode int
|
|
statusLine string
|
|
endOnce sync.Once
|
|
writer http.ResponseWriter
|
|
}
|
|
|
|
// Compile time assertions for widely used net/http interfaces
|
|
var _ http.CloseNotifier = (*trackingResponseWriter)(nil)
|
|
var _ http.Flusher = (*trackingResponseWriter)(nil)
|
|
var _ http.Hijacker = (*trackingResponseWriter)(nil)
|
|
var _ http.Pusher = (*trackingResponseWriter)(nil)
|
|
var _ http.ResponseWriter = (*trackingResponseWriter)(nil)
|
|
|
|
var errHijackerUnimplemented = errors.New("ResponseWriter does not implement http.Hijacker")
|
|
|
|
func (t *trackingResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
|
|
hj, ok := t.writer.(http.Hijacker)
|
|
if !ok {
|
|
return nil, nil, errHijackerUnimplemented
|
|
}
|
|
return hj.Hijack()
|
|
}
|
|
|
|
func (t *trackingResponseWriter) CloseNotify() <-chan bool {
|
|
cn, ok := t.writer.(http.CloseNotifier)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
return cn.CloseNotify()
|
|
}
|
|
|
|
func (t *trackingResponseWriter) Push(target string, opts *http.PushOptions) error {
|
|
pusher, ok := t.writer.(http.Pusher)
|
|
if !ok {
|
|
return http.ErrNotSupported
|
|
}
|
|
return pusher.Push(target, opts)
|
|
}
|
|
|
|
func (t *trackingResponseWriter) end() {
|
|
t.endOnce.Do(func() {
|
|
if t.statusCode == 0 {
|
|
t.statusCode = 200
|
|
}
|
|
|
|
span := trace.FromContext(t.ctx)
|
|
span.SetStatus(TraceStatus(t.statusCode, t.statusLine))
|
|
|
|
m := []stats.Measurement{
|
|
ServerLatency.M(float64(time.Since(t.start)) / float64(time.Millisecond)),
|
|
ServerResponseBytes.M(t.respSize),
|
|
}
|
|
if t.reqSize >= 0 {
|
|
m = append(m, ServerRequestBytes.M(t.reqSize))
|
|
}
|
|
ctx, _ := tag.New(t.ctx, tag.Upsert(StatusCode, strconv.Itoa(t.statusCode)))
|
|
stats.Record(ctx, m...)
|
|
})
|
|
}
|
|
|
|
func (t *trackingResponseWriter) Header() http.Header {
|
|
return t.writer.Header()
|
|
}
|
|
|
|
func (t *trackingResponseWriter) Write(data []byte) (int, error) {
|
|
n, err := t.writer.Write(data)
|
|
t.respSize += int64(n)
|
|
return n, err
|
|
}
|
|
|
|
func (t *trackingResponseWriter) WriteHeader(statusCode int) {
|
|
t.writer.WriteHeader(statusCode)
|
|
t.statusCode = statusCode
|
|
t.statusLine = http.StatusText(t.statusCode)
|
|
}
|
|
|
|
func (t *trackingResponseWriter) Flush() {
|
|
if flusher, ok := t.writer.(http.Flusher); ok {
|
|
flusher.Flush()
|
|
}
|
|
}
|