Skip to content
This repository was archived by the owner on Aug 23, 2023. It is now read-only.

Commit 61ea62a

Browse files
DieterbeDieterbe
authored
Merge pull request #1161 from grafana/notifier-stuff
notifier: deprecate NSQ ( rip :'( ) + add mt-kafka-persist-sniff tool
2 parents 38d5118 + 048e1d1 commit 61ea62a

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

51 files changed

+174
-5882
lines changed

‎.gitignore

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
/cmd/mt-index-migrate/mt-index-migrate
99
/cmd/mt-kafka-mdm-sniff-out-of-order/mt-kafka-mdm-sniff-out-of-order
1010
/cmd/mt-kafka-mdm-sniff/mt-kafka-mdm-sniff
11+
/cmd/mt-kafka-persist-sniff/mt-kafka-persist-sniff
1112
/cmd/mt-schemas-explain/mt-schemas-explain
1213
/cmd/mt-split-metrics-by-ttl/mt-split-metrics-by-ttl
1314
/cmd/mt-store-cat/mt-store-cat

‎Gopkg.lock

-37
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Gopkg.toml

-12
Original file line numberDiff line numberDiff line change
@@ -45,10 +45,6 @@ unused-packages = true
4545
name = "github.com/alyu/configparser"
4646
branch = "master"
4747

48-
[[constraint]]
49-
name = "github.com/bitly/go-hostpool"
50-
branch = "master"
51-
5248
[[constraint]]
5349
name = "github.com/davecgh/go-spew"
5450
version = "1.1.0"
@@ -97,10 +93,6 @@ unused-packages = true
9793
name = "github.com/mitchellh/go-homedir"
9894
branch = "master"
9995

100-
[[constraint]]
101-
name = "github.com/nsqio/go-nsq"
102-
revision = "642a3f9935f12cb3b747294318d730f56f4c34b4"
103-
10496
[[constraint]]
10597
name = "github.com/opentracing/opentracing-go"
10698
version = "^1"
@@ -117,10 +109,6 @@ unused-packages = true
117109
name = "github.com/raintank/met"
118110
branch = "master"
119111

120-
[[constraint]]
121-
name = "github.com/raintank/misc"
122-
branch = "master"
123-
124112
[[constraint]]
125113
name = "github.com/raintank/schema"
126114
version = "^2"

‎cmd/metrictank/metrictank.go

+4-13
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@ import (
3333
"github.com/grafana/metrictank/mdata"
3434
"github.com/grafana/metrictank/mdata/cache"
3535
"github.com/grafana/metrictank/mdata/notifierKafka"
36-
"github.com/grafana/metrictank/mdata/notifierNsq"
3736
"github.com/grafana/metrictank/stats"
3837
statsConfig "github.com/grafana/metrictank/stats/config"
3938
bigtableStore "github.com/grafana/metrictank/store/bigtable"
@@ -112,9 +111,6 @@ func main() {
112111
inKafkaMdm.ConfigSetup()
113112
inPrometheus.ConfigSetup()
114113

115-
// load config for cluster handlers
116-
notifierNsq.ConfigSetup()
117-
118114
// load config for metricIndexers
119115
memory.ConfigSetup()
120116
cassandra.ConfigSetup()
@@ -184,7 +180,6 @@ func main() {
184180
inKafkaMdm.ConfigProcess(*instance)
185181
memory.ConfigProcess()
186182
inPrometheus.ConfigProcess()
187-
notifierNsq.ConfigProcess()
188183
notifierKafka.ConfigProcess(*instance)
189184
statsConfig.ConfigProcess(*instance)
190185
mdata.ConfigProcess()
@@ -387,18 +382,14 @@ func main() {
387382
/***********************************
388383
Initialize MetricPersist notifiers
389384
***********************************/
390-
handlers := make([]mdata.NotifierHandler, 0)
385+
var notifiers []mdata.Notifier
391386
if notifierKafka.Enabled {
392-
// The notifierKafka handler will block here until it has processed the backlog of metricPersist messages.
387+
// The notifierKafka notifiers will block here until it has processed the backlog of metricPersist messages.
393388
// it will block for at most kafka-cluster.backlog-process-timeout (default 60s)
394-
handlers = append(handlers, notifierKafka.New(*instance, metrics, metricIndex))
395-
}
396-
397-
if notifierNsq.Enabled {
398-
handlers = append(handlers, notifierNsq.New(*instance, metrics, metricIndex))
389+
notifiers = append(notifiers, notifierKafka.New(*instance, mdata.NewDefaultNotifierHandler(metrics, metricIndex)))
399390
}
400391

401-
mdata.InitPersistNotifier(handlers...)
392+
mdata.InitPersistNotifier(notifiers...)
402393

403394
/***********************************
404395
Start our inputs

‎cmd/mt-kafka-persist-sniff/handler.go

+38
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package main
2+
3+
import (
4+
"encoding/json"
5+
"fmt"
6+
7+
"github.com/grafana/metrictank/mdata"
8+
"github.com/raintank/schema"
9+
log "github.com/sirupsen/logrus"
10+
)
11+
12+
type PrintNotifierHandler struct{}
13+
14+
func NewPrintNotifierHandler() PrintNotifierHandler {
15+
return PrintNotifierHandler{}
16+
}
17+
18+
func (dn PrintNotifierHandler) PartitionOf(key schema.MKey) (int32, bool) {
19+
return 0, false
20+
}
21+
22+
func (dn PrintNotifierHandler) Handle(data []byte) {
23+
version := uint8(data[0])
24+
if version == uint8(mdata.PersistMessageBatchV1) {
25+
batch := mdata.PersistMessageBatch{}
26+
err := json.Unmarshal(data[1:], &batch)
27+
if err != nil {
28+
log.Errorf("failed to unmarsh batch message: %s -- skipping", err)
29+
return
30+
}
31+
for _, c := range batch.SavedChunks {
32+
fmt.Printf("%s %d %s\n", batch.Instance, c.T0, c.Key)
33+
}
34+
} else {
35+
log.Errorf("unknown message version %d", version)
36+
}
37+
return
38+
}

‎cmd/mt-kafka-persist-sniff/main.go

+55
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package main
2+
3+
import (
4+
"flag"
5+
"fmt"
6+
"math/rand"
7+
"os"
8+
"os/signal"
9+
"strconv"
10+
"syscall"
11+
12+
"github.com/grafana/metrictank/logger"
13+
"github.com/grafana/metrictank/mdata/notifierKafka"
14+
"github.com/grafana/metrictank/stats"
15+
log "github.com/sirupsen/logrus"
16+
)
17+
18+
func main() {
19+
flag.Usage = func() {
20+
fmt.Fprintln(os.Stderr, "mt-kafka-persist-sniff")
21+
fmt.Fprintln(os.Stderr, "Print what's flowing through kafka metric persist topic")
22+
fmt.Fprintf(os.Stderr, "\nFlags:\n\n")
23+
flag.PrintDefaults()
24+
notifierKafka.FlagSet.PrintDefaults()
25+
}
26+
formatter := &logger.TextFormatter{}
27+
formatter.TimestampFormat = "2006-01-02 15:04:05.000"
28+
log.SetFormatter(formatter)
29+
log.SetLevel(log.InfoLevel)
30+
instance := "mt-kafka-persist-sniff" + strconv.Itoa(rand.Int())
31+
32+
notifierKafka.FlagSet.Usage = flag.Usage
33+
notifierKafka.FlagSet.Parse(os.Args[1:])
34+
// config may have had it disabled
35+
notifierKafka.Enabled = true
36+
37+
stats.NewDevnull() // make sure metrics don't pile up without getting discarded
38+
39+
notifierKafka.ConfigProcess(instance)
40+
41+
done := make(chan struct{})
42+
go func() {
43+
notifierKafka.New(instance, NewPrintNotifierHandler())
44+
close(done)
45+
}()
46+
sigChan := make(chan os.Signal, 1)
47+
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
48+
49+
select {
50+
case sig := <-sigChan:
51+
log.Infof("Received signal %q. Shutting down", sig)
52+
case <-done:
53+
}
54+
// notifierKafka.Stop()
55+
}

‎dashboards/extra/fake-metrics.json

-92
Original file line numberDiff line numberDiff line change
@@ -535,98 +535,6 @@
535535
],
536536
"title": "New row"
537537
},
538-
{
539-
"collapse": false,
540-
"editable": true,
541-
"height": "250px",
542-
"panels": [
543-
{
544-
"aliasColors": {
545-
"timeout_count": "#890F02"
546-
},
547-
"bars": false,
548-
"datasource": "graphite",
549-
"editable": true,
550-
"error": false,
551-
"fill": 1,
552-
"grid": {
553-
"threshold1": null,
554-
"threshold1Color": "rgba(216, 200, 27, 0.27)",
555-
"threshold2": null,
556-
"threshold2Color": "rgba(234, 112, 112, 0.22)"
557-
},
558-
"id": 4,
559-
"isNew": true,
560-
"legend": {
561-
"avg": false,
562-
"current": false,
563-
"max": false,
564-
"min": false,
565-
"show": true,
566-
"total": false,
567-
"values": false
568-
},
569-
"lines": true,
570-
"linewidth": 2,
571-
"links": [],
572-
"nullPointMode": "connected",
573-
"percentage": false,
574-
"pointradius": 5,
575-
"points": false,
576-
"renderer": "flot",
577-
"seriesOverrides": [
578-
{
579-
"alias": "depth",
580-
"fill": 3,
581-
"linewidth": 0,
582-
"yaxis": 2
583-
}
584-
],
585-
"span": 12,
586-
"stack": false,
587-
"steppedLine": false,
588-
"targets": [
589-
{
590-
"refId": "A",
591-
"target": "aliasByNode(stats.$environment.nsq.*.topic.metrics.channel.tank.*, 8)"
592-
},
593-
{
594-
"refId": "B",
595-
"target": "aliasByNode(stats.$environment.gauges.nsq.*.topic.metrics.depth, 7)"
596-
}
597-
],
598-
"timeFrom": null,
599-
"timeShift": null,
600-
"title": "NSQ topic/channel",
601-
"tooltip": {
602-
"shared": true,
603-
"value_type": "cumulative",
604-
"msResolution": false
605-
},
606-
"type": "graph",
607-
"yaxes": [
608-
{
609-
"show": true,
610-
"min": null,
611-
"max": null,
612-
"logBase": 1,
613-
"format": "short"
614-
},
615-
{
616-
"show": true,
617-
"min": null,
618-
"max": null,
619-
"logBase": 1,
620-
"format": "short"
621-
}
622-
],
623-
"xaxis": {
624-
"show": true
625-
}
626-
}
627-
],
628-
"title": "New row"
629-
},
630538
{
631539
"collapse": false,
632540
"editable": true,

‎docker/docker-chaos/metrictank.ini

-17
Original file line numberDiff line numberDiff line change
@@ -332,23 +332,6 @@ offset = newest
332332
# Maximum time backlog processing can block during metrictank startup.
333333
backlog-process-timeout = 60s
334334

335-
### nsq as transport for clustering messages
336-
[nsq-cluster]
337-
enabled = false
338-
# nsqd TCP address (may be given multiple times as comma-separated list)
339-
nsqd-tcp-address =
340-
# lookupd HTTP address (may be given multiple times as comma-separated list)
341-
lookupd-http-address =
342-
topic = metricpersist
343-
channel = tank
344-
# passthrough to nsq.Producer (may be given multiple times as comma-separated list, see http://godoc.org/github.com/nsqio/go-nsq#Config)")
345-
producer-opt =
346-
#passthrough to nsq.Consumer (may be given multiple times as comma-separated list, http://godoc.org/github.com/nsqio/go-nsq#Config)")
347-
consumer-opt =
348-
# max number of messages to allow in flight
349-
max-in-flight = 200
350-
351-
352335
## metric metadata index ##
353336

354337
### in memory, cassandra-backed

0 commit comments

Comments
 (0)