list dropped handler

README.md

@@ -179,7 +179,16 @@ enabled = false
 drop-future = "0s"
 drop-past = "0s"
 
-[pprof]
+# Golang pprof + some extra locations
+#
+# Last 1000 points dropped by "drop-future" and "drop-past" rules:
+# /debug/receive/tcp/dropped/
+# /debug/receive/udp/dropped/
+# /debug/receive/pickle/dropped/
+# /debug/receive/grpc/dropped/
+# /debug/receive/prometheus/dropped/
+# /debug/receive/telegraf_http_json/dropped/
+[pprof] 
 listen = "localhost:7007"
 enabled = false
 ```

carbon/app.go

@@ -242,6 +242,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/tcp/dropped/", app.TCP.DroppedHandler)
 	}
 
 	if conf.Udp.Enabled {
@@ -256,6 +258,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/udp/dropped/", app.UDP.DroppedHandler)
 	}
 
 	if conf.Pickle.Enabled {
@@ -270,6 +274,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/pickle/dropped/", app.Pickle.DroppedHandler)
 	}
 
 	if conf.Grpc.Enabled {
@@ -283,6 +289,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/grpc/dropped/", app.Grpc.DroppedHandler)
 	}
 
 	if conf.Prometheus.Enabled {
@@ -296,6 +304,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/prometheus/dropped/", app.Prometheus.DroppedHandler)
 	}
 
 	if conf.TelegrafHttpJson.Enabled {
@@ -309,6 +319,8 @@ func (app *App) Start() (err error) {
 		if err != nil {
 			return
 		}
+
+		http.HandleFunc("/debug/receive/telegraf_http_json/dropped/", app.TelegrafHttpJson.DroppedHandler)
 	}
 	/* RECEIVER end */
 

receiver/base.go

@@ -1,13 +1,19 @@
 package receiver
 
 import (
+	"fmt"
+	"net/http"
+	"sort"
+	"sync"
 	"sync/atomic"
 
 	"github.com/lomik/carbon-clickhouse/helper/RowBinary"
 	"github.com/lomik/carbon-clickhouse/helper/stop"
 	"go.uber.org/zap"
 )
 
+const droppedListSize = 1000
+
 type Base struct {
 	stop.Struct
 	stat struct {
@@ -20,6 +26,9 @@ type Base struct {
 		futureDropped      uint64 // atomic
 		pastDropped        uint64 // atomic
 	}
+	droppedList       [droppedListSize]string
+	droppedListNext   int
+	droppedListMu     sync.Mutex
 	parseThreads      int
 	dropFutureSeconds uint32
 	dropPastSeconds   uint32
@@ -53,6 +62,46 @@ func (base *Base) isDrop(nowTime uint32, metricTime uint32) bool {
 	return false
 }
 
+func (base *Base) DroppedHandler(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "text/plain")
+
+	n := make([]string, droppedListSize)
+	base.droppedListMu.Lock()
+	copy(n, base.droppedList[:])
+	base.droppedListMu.Unlock()
+	sort.Strings(n)
+	for i := 0; i < len(n); i++ {
+		if n[i] != "" {
+			fmt.Fprintln(w, n[i])
+		}
+	}
+}
+
+func (base *Base) saveDropped(name string, nowTime uint32, metricTime uint32, value float64) {
+	s := fmt.Sprintf("rcv:%d\tname:%s\ttimestamp:%d\tvalue:%#v", nowTime, name, metricTime, value)
+
+	base.droppedListMu.Lock()
+	base.droppedList[base.droppedListNext%droppedListSize] = s
+	base.droppedListNext++
+	base.droppedListMu.Unlock()
+}
+
+func (base *Base) isDropString(name string, nowTime uint32, metricTime uint32, value float64) bool {
+	if !base.isDrop(nowTime, metricTime) {
+		return false
+	}
+	base.saveDropped(name, nowTime, metricTime, value)
+	return true
+}
+
+func (base *Base) isDropBytes(name []byte, nowTime uint32, metricTime uint32, value float64) bool {
+	if !base.isDrop(nowTime, metricTime) {
+		return false
+	}
+	base.saveDropped(string(name), nowTime, metricTime, value)
+	return true
+}
+
 func (base *Base) SendStat(send func(metric string, value float64), fields ...string) {
 	for _, f := range fields {
 		switch f {

receiver/grpc.go

@@ -131,7 +131,7 @@ func (g *GRPC) doStore(requestCtx context.Context, in *pb.Payload, confirmRequir
 		m := in.Metrics[i]
 
 		for j := 0; j < len(m.Points); j++ {
-			if g.isDrop(now, m.Points[j].Timestamp) {
+			if g.isDropString(m.Metric, now, m.Points[j].Timestamp, m.Points[j].Value) {
 				continue
 			}
 

receiver/pickle_parser.go

@@ -48,16 +48,16 @@ func (base *Base) PickleParseBytes(ctx context.Context, b []byte, now uint32) {
 	}
 
 	pickle.ParseMessage(b, func(name string, value float64, timestamp int64) {
-		if base.isDrop(now, uint32(timestamp)) {
-			return
-		}
-
 		name, err := tags.Graphite(name)
 		if err != nil {
 			// @TODO: log?
 			return
 		}
 
+		if base.isDropString(name, now, uint32(timestamp), value) {
+			return
+		}
+
 		if !wb.CanWriteGraphitePoint(len(name)) {
 			flush()
 			if len(name) > RowBinary.WriteBufferSize-50 {

receiver/plain.go

@@ -120,7 +120,7 @@ MainLoop:
 			continue MainLoop
 		}
 
-		if base.isDrop(b.Time, timestamp) {
+		if base.isDropBytes(name, b.Time, timestamp, value) {
 			continue MainLoop
 		}
 

receiver/prometheus.go

@@ -62,7 +62,7 @@ func (rcv *PrometheusRemoteWrite) ServeHTTP(w http.ResponseWriter, r *http.Reque
 			if math.IsNaN(samples[j].Value) {
 				continue
 			}
-			if rcv.isDrop(writer.Now(), uint32(samples[j].Timestamp/1000)) {
+			if rcv.isDropString(metric, writer.Now(), uint32(samples[j].Timestamp/1000), samples[j].Value) {
 				continue
 			}
 

receiver/receiver.go

@@ -3,6 +3,7 @@ package receiver
 import (
 	"fmt"
 	"net"
+	"net/http"
 	"net/url"
 	"strings"
 
@@ -12,6 +13,7 @@ import (
 
 type Receiver interface {
 	Stat(func(metric string, value float64))
+	DroppedHandler(w http.ResponseWriter, r *http.Request)
 	Stop()
 }
 

receiver/telegraf_http_json.go

@@ -91,11 +91,9 @@ func (rcv *TelegrafHttpJson) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	var pathBuf bytes.Buffer
 
+metricsLoop:
 	for i := 0; i < len(data.Metrics); i++ {
 		m := data.Metrics[i]
-		if rcv.isDrop(writer.Now(), uint32(m.Timestamp)) {
-			continue
-		}
 
 		tags := TelegrafEncodeTags(m.Tags)
 
@@ -127,7 +125,11 @@ func (rcv *TelegrafHttpJson) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			pathBuf.WriteByte('?')
 			pathBuf.WriteString(tags)
 
-			writer.WritePoint(pathBuf.String(), v, m.Timestamp)
+			name := pathBuf.String()
+			if rcv.isDropString(name, writer.Now(), uint32(m.Timestamp), v) {
+				continue metricsLoop
+			}
+			writer.WritePoint(name, v, m.Timestamp)
 		}
 	}