// NewClientStream creates a new Stream for the client side. This is called
// by generated code.
func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
	var (
		t   transport.ClientTransport
		err error
	)
	t, err = cc.dopts.picker.Pick(ctx)
	if err != nil {
		return nil, toRPCErr(err)
	}
	// TODO(zhaoq): CallOption is omitted. Add support when it is needed.
	callHdr := &transport.CallHdr{
		Host:   cc.authority,
		Method: method,
		Flush:  desc.ServerStreams && desc.ClientStreams,
	}
	if cc.dopts.cp != nil {
		callHdr.SendCompress = cc.dopts.cp.Type()
	}
	cs := &clientStream{
		desc:    desc,
		codec:   cc.dopts.codec,
		cp:      cc.dopts.cp,
		dc:      cc.dopts.dc,
		tracing: EnableTracing,
	}
	if cc.dopts.cp != nil {
		callHdr.SendCompress = cc.dopts.cp.Type()
		cs.cbuf = new(bytes.Buffer)
	}
	if cs.tracing {
		cs.trInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
		cs.trInfo.firstLine.client = true
		if deadline, ok := ctx.Deadline(); ok {
			cs.trInfo.firstLine.deadline = deadline.Sub(time.Now())
		}
		cs.trInfo.tr.LazyLog(&cs.trInfo.firstLine, false)
		ctx = trace.NewContext(ctx, cs.trInfo.tr)
	}
	s, err := t.NewStream(ctx, callHdr)
	if err != nil {
		cs.finish(err)
		return nil, toRPCErr(err)
	}
	cs.t = t
	cs.s = s
	cs.p = &parser{r: s}
	// Listen on ctx.Done() to detect cancellation when there is no pending
	// I/O operations on this stream.
	go func() {
		select {
		case <-t.Error():
			// Incur transport error, simply exit.
		case <-s.Context().Done():
			err := s.Context().Err()
			cs.finish(err)
			cs.closeTransportStream(transport.ContextErr(err))
		}
	}()
	return cs, nil
}

func (bp *Backplane) handleStats(w http.ResponseWriter, req *http.Request) {
	tr := trace.New("backend.internalstats", req.RequestURI)
	defer tr.Finish()
	hostname, err := os.Hostname()
	if err != nil {
		glog.Error("Unable to obtain hostname: ", err)
		tr.LazyPrintf("Unable to obtain hostname: ", err)
	}
	var data = struct {
		Backends                         []*Backend
		Frontends                        []*Frontend
		Pid                              int
		Hostname                         string
		Uptime                           time.Duration
		LimitAs, LimitFsize, LimitNofile syscall.Rlimit
	}{
		Backends:  bp.Backends,
		Frontends: bp.Frontends,
		Pid:       os.Getpid(),
		Hostname:  hostname,
		Uptime:    time.Since(starttime),
	}
	syscall.Getrlimit(syscall.RLIMIT_AS, &data.LimitAs)
	syscall.Getrlimit(syscall.RLIMIT_FSIZE, &data.LimitFsize)
	syscall.Getrlimit(syscall.RLIMIT_NOFILE, &data.LimitNofile)

	err = StatsTemplate().Execute(w, &data)
	if err != nil {
		glog.Errorf("unable to execute template: %s", err)
		tr.LazyPrintf("unable to execute template: %s", err)
	}
}

// NewSession creates and initializes new Session object.
// ctx can be nil (in which case the Executor's context will be used).
// remote can be nil.
func NewSession(ctx context.Context, args SessionArgs, e *Executor, remote net.Addr) *Session {
	s := &Session{
		Database: args.Database,
		User:     args.User,
		Location: time.UTC,
	}
	cfg, cache := e.getSystemConfig()
	s.planner = planner{
		leaseMgr:      e.ctx.LeaseManager,
		systemConfig:  cfg,
		databaseCache: cache,
		session:       s,
		execCtx:       &e.ctx,
	}
	s.PreparedStatements = makePreparedStatements(s)
	s.PreparedPortals = makePreparedPortals(s)
	remoteStr := ""
	if remote != nil {
		remoteStr = remote.String()
	}
	s.Trace = trace.New("sql."+args.User, remoteStr)
	s.Trace.SetMaxEvents(100)
	s.context, s.cancel = context.WithCancel(ctx)
	return s
}

// netTraceIntegrator is passed into basictracer as NewSpanEventListener
// and causes all traces to be registered with the net/trace endpoint.
func netTraceIntegrator() func(basictracer.SpanEvent) {
	var tr trace.Trace
	return func(e basictracer.SpanEvent) {
		switch t := e.(type) {
		case basictracer.EventCreate:
			tr = trace.New("tracing", t.OperationName)
			tr.SetMaxEvents(maxLogsPerSpan)
		case basictracer.EventFinish:
			tr.Finish()
		case basictracer.EventTag:
			tr.LazyPrintf("%s:%v", t.Key, t.Value)
		case basictracer.EventLogFields:
			// TODO(radu): when LightStep supports arbitrary fields, we should make
			// the formatting of the message consistent with that. Until then we treat
			// legacy events that just have an "event" key specially.
			if len(t.Fields) == 1 && t.Fields[0].Key() == "event" {
				tr.LazyPrintf("%s", t.Fields[0].Value())
			} else {
				var buf bytes.Buffer
				for i, f := range t.Fields {
					if i > 0 {
						buf.WriteByte(' ')
					}
					fmt.Fprintf(&buf, "%s:%v", f.Key(), f.Value())
				}

				tr.LazyPrintf("%s", buf.String())
			}
		case basictracer.EventLog:
			panic("EventLog is deprecated")
		}
	}
}

func (s *CountersCollectingRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
	tr := trace.New(s.TraceFamily, r.RequestURI)
	tr.LazyPrintf("Request: %#v", r)
	defer tr.Finish()
	if s.RateLimiter != nil {
		if !s.RateLimiter.Accepted() {
			tr.LazyPrintf("Rate limited")
			tr.SetError()
			return nil, RateLimited
		}
	}
	s.stats.in()
	resp, err := s.RoundTripper.RoundTrip(r)
	s.stats.out()
	if resp != nil && err == nil {
		respBucket := resp.StatusCode / 100
		if respBucket == 0 {
			respBucket = 2
		}
		if respBucket > 0 && respBucket <= 5 {
			atomic.AddInt64(&s.stats.CountersByResponseCode[respBucket], 1)
		}
	}
	if err != nil {
		tr.LazyPrintf("Error in roundtripper: %s", err)
		tr.SetError()
	}
	tr.LazyPrintf("Response: %v", resp)
	return resp, err
}

func (ntwe *netTraceWrapExtractor) JoinTrace(opName string, carrier interface{}) (opentracing.Span, error) {
	sp, err := ntwe.wrap.JoinTrace(opName, carrier)
	if err != nil {
		return nil, err
	}
	return &netTraceWrapSpan{Span: sp, tr: trace.New(family, opName)}, nil
}

// Common handler for both javascript responses.
func (s *Server) HTTPResponse(w http.ResponseWriter, r *http.Request) {
	tr := trace.New("http", "response")
	defer tr.Finish()

	if !s.ratelimiter.Allowed() {
		tr.LazyPrintf("rate limited")
		tr.SetError()
		http.Error(w, "too many requests", 429)
		return
	}

	key := keyFromRequest(r)
	tr.LazyPrintf("key: %s", key)
	op, ok := s.getOp(key)
	if !ok {
		tr.LazyPrintf("404 error")
		tr.SetError()
		http.NotFound(w, r)
		return
	}

	buf := make([]byte, 4*1024)
	n, err := r.Body.Read(buf)

	if err != nil && err != io.EOF {
		tr.LazyPrintf("400 error reading body: %v", err)
		tr.SetError()
		http.Error(w, "error reading body", http.StatusBadRequest)
		return
	}

	op.reply <- buf[:n]

	w.Write([]byte("success"))
}

// NewClientStream creates a new Stream for the client side. This is called
// by generated code.
func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
	// TODO(zhaoq): CallOption is omitted. Add support when it is needed.
	callHdr := &transport.CallHdr{
		Host:   cc.authority,
		Method: method,
	}
	cs := &clientStream{
		desc:    desc,
		codec:   cc.dopts.codec,
		tracing: EnableTracing,
	}
	if cs.tracing {
		cs.traceInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
		cs.traceInfo.firstLine.client = true
		if deadline, ok := ctx.Deadline(); ok {
			cs.traceInfo.firstLine.deadline = deadline.Sub(time.Now())
		}
		cs.traceInfo.tr.LazyLog(&cs.traceInfo.firstLine, false)
	}
	t, err := cc.wait(ctx)
	if err != nil {
		return nil, toRPCErr(err)
	}
	s, err := t.NewStream(ctx, callHdr)
	if err != nil {
		return nil, toRPCErr(err)
	}
	cs.t = t
	cs.s = s
	cs.p = &parser{s: s}
	return cs, nil
}

// netTraceIntegrator is passed into basictracer as NewSpanEventListener
// and causes all traces to be registered with the net/trace endpoint.
func netTraceIntegrator() func(basictracer.SpanEvent) {
	var tr trace.Trace
	return func(e basictracer.SpanEvent) {
		switch t := e.(type) {
		case basictracer.EventCreate:
			tr = trace.New("tracing", t.OperationName)
			tr.SetMaxEvents(maxLogsPerSpan)
		case basictracer.EventFinish:
			tr.Finish()
		case basictracer.EventTag:
			tr.LazyPrintf("%s:%v", t.Key, t.Value)
		case basictracer.EventLogFields:
			var buf bytes.Buffer
			for i, f := range t.Fields {
				if i > 0 {
					buf.WriteByte(' ')
				}
				fmt.Fprintf(&buf, "%s:%v", f.Key(), f.Value())
			}

			tr.LazyPrintf("%s", buf.String())
		case basictracer.EventLog:
			if t.Payload != nil {
				tr.LazyPrintf("%s (payload %v)", t.Event, t.Payload)
			} else {
				tr.LazyPrintf("%s", t.Event)
			}
		}
	}
}

func (c *cachingResolver) Maintain() {
	go c.back.Maintain()

	for range time.Tick(maintenancePeriod) {
		tr := trace.New("dnstox.Cache", "GC")
		var total, expired int

		c.mu.Lock()
		total = len(c.answer)
		for q, ans := range c.answer {
			newTTL := getTTL(ans) - maintenancePeriod
			if newTTL > 0 {
				// Don't modify in place, create a copy and override.
				// That way, we avoid races with users that have gotten a
				// cached answer and are returning it.
				newans := copyRRSlice(ans)
				setTTL(newans, newTTL)
				c.answer[q] = newans
				continue
			}

			delete(c.answer, q)
			expired++
		}
		c.mu.Unlock()
		tr.LazyPrintf("total: %d   expired: %d", total, expired)
		tr.Finish()
	}
}

func (c *v3Conn) parseOptions(data []byte) error {
	defer func() {
		c.session.Trace = trace.New("sql."+c.opts.user, c.conn.RemoteAddr().String())
		c.session.Trace.SetMaxEvents(100)
	}()

	buf := readBuffer{msg: data}
	for {
		key, err := buf.getString()
		if err != nil {
			return util.Errorf("error reading option key: %s", err)
		}
		if len(key) == 0 {
			return nil
		}
		value, err := buf.getString()
		if err != nil {
			return util.Errorf("error reading option value: %s", err)
		}
		switch key {
		case "database":
			c.session.Database = value
		case "user":
			c.opts.user = value
		default:
			if log.V(1) {
				log.Warningf("unrecognized configuration parameter %q", key)
			}
		}
	}
}

func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc) (err error) {
	ss := &serverStream{
		t:       t,
		s:       stream,
		p:       &parser{s: stream},
		codec:   s.opts.codec,
		tracing: EnableTracing,
	}
	if ss.tracing {
		ss.traceInfo.tr = trace.New("grpc.Recv."+methodFamily(stream.Method()), stream.Method())
		ss.traceInfo.firstLine.client = false
		ss.traceInfo.tr.LazyLog(&ss.traceInfo.firstLine, false)
		defer func() {
			ss.mu.Lock()
			if err != nil && err != io.EOF {
				ss.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
				ss.traceInfo.tr.SetError()
			}
			ss.traceInfo.tr.Finish()
			ss.traceInfo.tr = nil
			ss.mu.Unlock()
		}()
	}
	if appErr := sd.Handler(srv.server, ss); appErr != nil {
		if err, ok := appErr.(rpcError); ok {
			ss.statusCode = err.code
			ss.statusDesc = err.desc
		} else {
			ss.statusCode = convertCode(appErr)
			ss.statusDesc = appErr.Error()
		}
	}
	return t.WriteStatus(ss.s, ss.statusCode, ss.statusDesc)

}

// StartOperation creates a new LogEntry with the current time.
// Should be immediately followed by a deferred call to FinishOperation.
func StartOperation(name string, path string) *LogEntry {
	return &LogEntry{
		name:      name,
		path:      path,
		startTime: time.Now(),
		trace:     trace.New(name, path),
	}
}

// Open implements http.FileSystem.
func (t *traceFS) Open(path string) (http.File, error) {
	tr := trace.New(t.family, path)
	ctx := trace.NewContext(context.Background(), tr)
	f, err := t.FileSystem.Open(ctx, path)

	// TODO: Decide where this should be in general (requests can be on-going).
	tr.Finish()
	return f, err
}

func routeTracing(route Route, handler xhandler.HandlerC) xhandler.HandlerC {
	rs := route.String()
	return xhandler.HandlerFuncC(func(ctx context.Context, w http.ResponseWriter, r *http.Request) {
		tr := trace.New(rs, fmt.Sprintf("%s %s", r.Method, r.URL.Path))
		ctx = trace.NewContext(ctx, tr)
		handler.ServeHTTPC(ctx, w, r)
		tr.Finish()
	})
}

func StartFileOperation(name, path string, args string) *LogEntry {
	name = "File " + name
	return &LogEntry{
		name:      name,
		path:      path,
		args:      args,
		startTime: time.Now(),
		trace:     trace.New(name, args),
	}
}

func (t *Tracer) newTrace(family string, id, name string) *Trace {
	nt := ntrace.New(family, name)
	nt.SetMaxEvents(100)
	return &Trace{
		ID:     id,
		Name:   name,
		family: family,
		tracer: t,
		nTrace: nt,
	}
}

func (t *TraceFS) newTrace(ctx context.Context, name, ft string, args ...interface{}) (trace.Trace, context.Context) {
	argsFmt := fmt.Sprintf(ft, args...)
	v := fmt.Sprintf("%s-%s %s", t.Id, t.MountConfig.FSName, argsFmt)

	if ctx == nil {
		ctx = context.TODO()
	}

	r := trace.New(name, v)
	ctx = trace.NewContext(ctx, r)

	return r, ctx
}

func WrapRequest(req *http.Request, traceFamily string) *Request {
	host, _, err := net.SplitHostPort(req.RemoteAddr)
	if err != nil {
		host = req.RemoteAddr
	}

	prefix := fmt.Sprintf("%s - %s - ", host, traceFamily)
	return &Request{
		Request: req,
		tr:      trace.New(traceFamily, req.URL.Path),
		log:     log.New(os.Stdout, prefix, log.LstdFlags),
	}
}

// traceInfo returns a traceInfo and associates it with stream, if tracing is enabled.
// If tracing is not enabled, it returns nil.
func (s *Server) traceInfo(st transport.ServerTransport, stream *transport.Stream) (trInfo *traceInfo) {
	if !EnableTracing {
		return nil
	}
	trInfo = &traceInfo{
		tr: trace.New("grpc.Recv."+methodFamily(stream.Method()), stream.Method()),
	}
	trInfo.firstLine.client = false
	trInfo.firstLine.remoteAddr = st.RemoteAddr()
	stream.TraceContext(trInfo.tr)
	if dl, ok := stream.Context().Deadline(); ok {
		trInfo.firstLine.deadline = dl.Sub(time.Now())
	}
	return trInfo
}