route_breakdown.go

package gobrake

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sync"
	"time"
)

type routeBreakdownKey struct {
	Method   string    `json:"method"`
	Route    string    `json:"route"`
	RespType string    `json:"responseType"`
	Time     time.Time `json:"time"`
}

type routeBreakdown struct {
	routeBreakdownKey
	tdigestStatGroups
}

type routeBreakdowns struct {
	opt        *NotifierOptions
	flushTimer *time.Timer
	addWG      *sync.WaitGroup

	mu sync.Mutex
	m  map[routeBreakdownKey]*routeBreakdown
}

func newRouteBreakdowns(opt *NotifierOptions) *routeBreakdowns {
	return &routeBreakdowns{
		opt: opt,
	}
}

func (s *routeBreakdowns) init() {
	if s.flushTimer == nil {
		s.flushTimer = time.AfterFunc(flushPeriod, s.Flush)
		s.addWG = new(sync.WaitGroup)
		s.m = make(map[routeBreakdownKey]*routeBreakdown)
	}
}

// Flush sends to Airbrake route stats.
func (s *routeBreakdowns) Flush() {
	s.mu.Lock()

	s.flushTimer = nil
	addWG := s.addWG
	s.addWG = nil
	m := s.m
	s.m = nil

	s.mu.Unlock()

	if m == nil {
		return
	}

	addWG.Wait()
	err := s.send(m)
	if err != nil {
		logger.Printf("routeBreakdowns.send failed: %s", err)
	}
}

type breakdownsOut struct {
	Env    string            `json:"environment"`
	Routes []*routeBreakdown `json:"routes"`
}

func (s *routeBreakdowns) send(m map[routeBreakdownKey]*routeBreakdown) error {
	var routes []*routeBreakdown
	for _, v := range m {
		err := v.Pack()
		if err != nil {
			return err
		}
		routes = append(routes, v)
	}

	buf := buffers.Get().(*bytes.Buffer)
	defer buffers.Put(buf)
	buf.Reset()

	out := breakdownsOut{
		Env:    s.opt.Environment,
		Routes: routes,
	}
	err := json.NewEncoder(buf).Encode(out)
	if err != nil {
		return err
	}

	req, err := http.NewRequest(
		http.MethodPut,
		fmt.Sprintf("%s/api/v5/projects/%d/routes-breakdowns",
			s.opt.APMHost, s.opt.ProjectId),
		buf,
	)
	if err != nil {
		return err
	}

	req.Header.Set("Authorization", "Bearer "+s.opt.ProjectKey)
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("User-Agent", userAgent)
	resp, err := s.opt.HTTPClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	buf.Reset()
	_, err = buf.ReadFrom(resp.Body)
	if err != nil {
		return err
	}

	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return nil
	}

	switch resp.StatusCode {
	case http.StatusUnauthorized:
		return errUnauthorized
	case http.StatusBadRequest, http.StatusTooManyRequests:
		var sendResp sendResponse
		err = json.NewDecoder(buf).Decode(&sendResp)
		if err != nil {
			return err
		}
		return errors.New(sendResp.Message)
	case 404, 408, 409, 410, 500, 502, 504:
		setRouteBreakdownBacklog(out)
	}

	err = fmt.Errorf("got unexpected response status=%q", resp.Status)
	return err
}

func (s *routeBreakdowns) Notify(c context.Context, metric *RouteMetric) error {
	if s.opt.DisableAPM {
		return fmt.Errorf(
			"APM is disabled, route breakdown is not sent: %s %s (status %d)",
			metric.Method, metric.Route, metric.StatusCode,
		)
	}

	if metric.StatusCode < 200 || (metric.StatusCode >= 300 && metric.StatusCode < 400) {
		// ignore
		return nil
	}

	total, err := metric.duration()
	if err != nil {
		return err
	}

	key := routeBreakdownKey{
		Method:   metric.Method,
		Route:    metric.Route,
		RespType: metric.respType(),
		Time:     metric.startTime.UTC().Truncate(time.Minute),
	}

	s.mu.Lock()
	s.init()
	b, ok := s.m[key]
	if !ok {
		b = &routeBreakdown{
			routeBreakdownKey: key,
		}
		s.m[key] = b
	}
	addWG := s.addWG
	addWG.Add(1)
	s.mu.Unlock()

	groups := metric.flushGroups()
	b.Add(total, groups)
	addWG.Done()

	return nil
}