datadog.go 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203
  1. package datadog
  2. import (
  3. "context"
  4. "net"
  5. "net/http"
  6. "os"
  7. "sync"
  8. "time"
  9. "github.com/DataDog/datadog-go/v5/statsd"
  10. "github.com/felixge/httpsnoop"
  11. log "github.com/sirupsen/logrus"
  12. "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
  13. "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
  14. "github.com/imgproxy/imgproxy/v3/config"
  15. "github.com/imgproxy/imgproxy/v3/metrics/errformat"
  16. "github.com/imgproxy/imgproxy/v3/metrics/stats"
  17. "github.com/imgproxy/imgproxy/v3/version"
  18. )
  19. type spanCtxKey struct{}
  20. type GaugeFunc func() float64
  21. var (
  22. enabled bool
  23. enabledMetrics bool
  24. statsdClient *statsd.Client
  25. statsdClientStop chan struct{}
  26. gaugeFuncs = make(map[string]GaugeFunc)
  27. gaugeFuncsMutex sync.RWMutex
  28. )
  29. func Init() {
  30. if !config.DataDogEnable {
  31. return
  32. }
  33. name := os.Getenv("DD_SERVICE")
  34. if len(name) == 0 {
  35. name = "imgproxy"
  36. }
  37. tracer.Start(
  38. tracer.WithService(name),
  39. tracer.WithServiceVersion(version.Version()),
  40. tracer.WithLogger(dataDogLogger{}),
  41. )
  42. enabled = true
  43. statsdHost, statsdPort := os.Getenv("DD_AGENT_HOST"), os.Getenv("DD_DOGSTATSD_PORT")
  44. if len(statsdHost) == 0 {
  45. statsdHost = "localhost"
  46. }
  47. if len(statsdPort) == 0 {
  48. statsdPort = "8125"
  49. }
  50. if !config.DataDogEnableMetrics {
  51. return
  52. }
  53. var err error
  54. statsdClient, err = statsd.New(
  55. net.JoinHostPort(statsdHost, statsdPort),
  56. statsd.WithTags([]string{
  57. "service:" + name,
  58. "version:" + version.Version(),
  59. }),
  60. )
  61. if err == nil {
  62. statsdClientStop = make(chan struct{})
  63. enabledMetrics = true
  64. go runMetricsCollector()
  65. } else {
  66. log.Warnf("Can't initialize DogStatsD client: %s", err)
  67. }
  68. }
  69. func Stop() {
  70. if enabled {
  71. tracer.Stop()
  72. if statsdClient != nil {
  73. close(statsdClientStop)
  74. statsdClient.Close()
  75. }
  76. }
  77. }
  78. func Enabled() bool {
  79. return enabled
  80. }
  81. func StartRootSpan(ctx context.Context, rw http.ResponseWriter, r *http.Request) (context.Context, context.CancelFunc, http.ResponseWriter) {
  82. if !enabled {
  83. return ctx, func() {}, rw
  84. }
  85. span := tracer.StartSpan(
  86. "request",
  87. tracer.Measured(),
  88. tracer.SpanType("web"),
  89. tracer.Tag(ext.HTTPMethod, r.Method),
  90. tracer.Tag(ext.HTTPURL, r.RequestURI),
  91. )
  92. cancel := func() { span.Finish() }
  93. newRw := httpsnoop.Wrap(rw, httpsnoop.Hooks{
  94. WriteHeader: func(next httpsnoop.WriteHeaderFunc) httpsnoop.WriteHeaderFunc {
  95. return func(statusCode int) {
  96. span.SetTag(ext.HTTPCode, statusCode)
  97. next(statusCode)
  98. }
  99. },
  100. })
  101. return context.WithValue(ctx, spanCtxKey{}, span), cancel, newRw
  102. }
  103. func StartSpan(ctx context.Context, name string) context.CancelFunc {
  104. if !enabled {
  105. return func() {}
  106. }
  107. if rootSpan, ok := ctx.Value(spanCtxKey{}).(tracer.Span); ok {
  108. span := tracer.StartSpan(name, tracer.Measured(), tracer.ChildOf(rootSpan.Context()))
  109. return func() { span.Finish() }
  110. }
  111. return func() {}
  112. }
  113. func SendError(ctx context.Context, errType string, err error) {
  114. if !enabled {
  115. return
  116. }
  117. if rootSpan, ok := ctx.Value(spanCtxKey{}).(tracer.Span); ok {
  118. rootSpan.SetTag(ext.Error, err)
  119. rootSpan.SetTag(ext.ErrorType, errformat.FormatErrType(errType, err))
  120. }
  121. }
  122. func AddGaugeFunc(name string, f GaugeFunc) {
  123. gaugeFuncsMutex.Lock()
  124. defer gaugeFuncsMutex.Unlock()
  125. gaugeFuncs["imgproxy."+name] = f
  126. }
  127. func ObserveBufferSize(t string, size int) {
  128. if enabledMetrics {
  129. statsdClient.Histogram("imgproxy.buffer.size", float64(size), []string{"type:" + t}, 1)
  130. }
  131. }
  132. func SetBufferDefaultSize(t string, size int) {
  133. if enabledMetrics {
  134. statsdClient.Gauge("imgproxy.buffer.default_size", float64(size), []string{"type:" + t}, 1)
  135. }
  136. }
  137. func SetBufferMaxSize(t string, size int) {
  138. if enabledMetrics {
  139. statsdClient.Gauge("imgproxy.buffer.max_size", float64(size), []string{"type:" + t}, 1)
  140. }
  141. }
  142. func runMetricsCollector() {
  143. tick := time.NewTicker(10 * time.Second)
  144. defer tick.Stop()
  145. for {
  146. select {
  147. case <-tick.C:
  148. func() {
  149. gaugeFuncsMutex.RLock()
  150. defer gaugeFuncsMutex.RUnlock()
  151. for name, f := range gaugeFuncs {
  152. statsdClient.Gauge(name, f(), nil, 1)
  153. }
  154. }()
  155. statsdClient.Gauge("imgproxy.requests_in_progress", stats.RequestsInProgress(), nil, 1)
  156. statsdClient.Gauge("imgproxy.images_in_progress", stats.ImagesInProgress(), nil, 1)
  157. case <-statsdClientStop:
  158. return
  159. }
  160. }
  161. }
  162. type dataDogLogger struct {
  163. }
  164. func (l dataDogLogger) Log(msg string) {
  165. log.Info(msg)
  166. }