From bbb35db3dad123691af416a4b827e12dcc584965 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 15:12:23 +0800 Subject: [PATCH 1/6] fix fmt --- main.go | 13 ++++--------- srv.go | 3 +-- writer.go | 3 +-- 3 files changed, 6 insertions(+), 13 deletions(-) diff --git a/main.go b/main.go index 02e6b09..319dab2 100644 --- a/main.go +++ b/main.go @@ -3,6 +3,7 @@ package main import ( "flag" "fmt" + "log" "os" "time" ) @@ -31,8 +32,6 @@ var ( ) func main() { - excode := 0 - conf := parseFlags() if versionFlag { @@ -41,27 +40,23 @@ func main() { if VersionPrerelease != "" { fmt.Println("Version PreRelease:", VersionPrerelease) } - os.Exit(excode) + return } fmt.Println("Starting up..") srv, err := NewP2CServer(conf) if err != nil { - fmt.Printf("Error: could not create server: %s\n", err.Error()) - excode = 1 - os.Exit(excode) + log.Fatalf("Error: could not create server: %s\n", err.Error()) } err = srv.Start() if err != nil { - fmt.Printf("Error: http server returned error: %s\n", err.Error()) - excode = 1 + log.Fatalf("Error: http server returned error: %s\n", err.Error()) } fmt.Println("Shutting down..") srv.Shutdown() fmt.Println("Exiting..") - os.Exit(excode) } func parseFlags() *config { diff --git a/srv.go b/srv.go index 1a9918a..259a901 100644 --- a/srv.go +++ b/srv.go @@ -1,12 +1,11 @@ package main import ( + "fmt" "io/ioutil" "net/http" "time" - "fmt" - "github.com/golang/protobuf/proto" "github.com/golang/snappy" "github.com/prometheus/client_golang/prometheus" diff --git a/writer.go b/writer.go index bba4ea8..1127208 100644 --- a/writer.go +++ b/writer.go @@ -4,9 +4,8 @@ import ( "database/sql" "fmt" "sort" - "time" - "sync" + "time" "github.com/kshvakov/clickhouse" "github.com/prometheus/client_golang/prometheus" From 67241f58ea125efa68b57891bee855c9c6ae6e2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 16:17:20 +0800 Subject: [PATCH 2/6] add zap logger --- main.go | 20 ++++++++++++-------- reader.go | 27 +++++++++++++++------------ srv.go | 19 +++++++++++-------- writer.go | 22 +++++++++++++--------- 4 files changed, 51 insertions(+), 37 deletions(-) diff --git a/main.go b/main.go index 319dab2..4c4cbb4 100644 --- a/main.go +++ b/main.go @@ -3,9 +3,10 @@ package main import ( "flag" "fmt" - "log" "os" "time" + + "go.uber.org/zap" ) // a lot of this borrows directly from: @@ -43,20 +44,23 @@ func main() { return } - fmt.Println("Starting up..") + logger, _ := zap.NewProduction() + defer logger.Sync() // flushes buffer, if any + sugar := logger.Sugar() + + sugar.Info("Starting up..") - srv, err := NewP2CServer(conf) + srv, err := NewP2CServer(conf, sugar) if err != nil { - log.Fatalf("Error: could not create server: %s\n", err.Error()) + sugar.Fatalf("could not create server: %s\n", err.Error()) } err = srv.Start() if err != nil { - log.Fatalf("Error: http server returned error: %s\n", err.Error()) + sugar.Fatalf("http server returned error: %s\n", err.Error()) } - - fmt.Println("Shutting down..") + sugar.Info("Shutting down..") srv.Shutdown() - fmt.Println("Exiting..") + sugar.Info("Exiting..") } func parseFlags() *config { diff --git a/reader.go b/reader.go index 2715132..5e3113c 100644 --- a/reader.go +++ b/reader.go @@ -9,11 +9,13 @@ import ( "github.com/prometheus/common/model" "github.com/prometheus/prometheus/storage/remote" + "go.uber.org/zap" ) type p2cReader struct { - conf *config - db *sql.DB + conf *config + db *sql.DB + logger *zap.SugaredLogger } // getTimePeriod return select and where SQL chunks relating to the time period -or- error @@ -152,13 +154,14 @@ func (r *p2cReader) getSQL(query *remote.Query) (string, error) { return sql, nil } -func NewP2CReader(conf *config) (*p2cReader, error) { +func NewP2CReader(conf *config, sugar *zap.SugaredLogger) (*p2cReader, error) { var err error r := new(p2cReader) r.conf = conf + r.logger = sugar r.db, err = sql.Open("clickhouse", r.conf.ChDSN) if err != nil { - fmt.Printf("Error connecting to clickhouse: %s\n", err.Error()) + r.logger.With("P2CReader").Errorf("connecting to clickhouse: %s\n", err.Error()) return r, err } @@ -182,27 +185,27 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) rcount := 0 for _, q := range req.Queries { // remove me.. - fmt.Printf("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) + r.logger.With("P2CReader").Info("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) // get the select sql sqlStr, err = r.getSQL(q) - fmt.Printf("query: running sql: %s\n\n", sqlStr) + r.logger.With("P2CReader").Info("query: running sql: %s\n\n", sqlStr) if err != nil { - fmt.Printf("Error: reader: getSQL: %s\n", err.Error()) + r.logger.With("P2CReader").Errorf("reader: getSQL: %s\n", err.Error()) return &resp, err } // get the select sql if err != nil { - fmt.Printf("Error: reader: getSQL: %s\n", err.Error()) + r.logger.With("P2CReader").Errorf("reader: getSQL: %s\n", err.Error()) return &resp, err } // todo: metrics on number of errors, rows, selects, timings, etc rows, err = r.db.Query(sqlStr) if err != nil { - fmt.Printf("Error: query failed: %s", sqlStr) - fmt.Printf("Error: query error: %s\n", err) + r.logger.With("P2CReader").Errorf("query failed: %s", sqlStr) + r.logger.With("P2CReader").Errorf("query error: %s\n", err) return &resp, err } @@ -218,7 +221,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) value float64 ) if err = rows.Scan(&cnt, &t, &name, &tags, &value); err != nil { - fmt.Printf("Error: scan: %s\n", err.Error()) + r.logger.With("P2CReader").Errorf("scan: %s\n", err.Error()) } // remove this.. //fmt.Printf(fmt.Sprintf("%d,%d,%s,%s,%f\n", cnt, t, name, strings.Join(tags, ":"), value)) @@ -244,7 +247,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) resp.Results[0].Timeseries = append(resp.Results[0].Timeseries, ts) } - fmt.Printf("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) + r.logger.With("P2CReader").Info("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) return &resp, nil diff --git a/srv.go b/srv.go index 259a901..efb6c61 100644 --- a/srv.go +++ b/srv.go @@ -11,6 +11,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/model" "github.com/prometheus/prometheus/storage/remote" + "go.uber.org/zap" "gopkg.in/tylerb/graceful.v1" ) @@ -28,24 +29,26 @@ type p2cServer struct { writer *p2cWriter reader *p2cReader rx prometheus.Counter + logger *zap.SugaredLogger } -func NewP2CServer(conf *config) (*p2cServer, error) { +func NewP2CServer(conf *config, sugar *zap.SugaredLogger) (*p2cServer, error) { var err error c := new(p2cServer) c.requests = make(chan *p2cRequest, conf.ChanSize) c.mux = http.NewServeMux() c.conf = conf + c.logger = sugar - c.writer, err = NewP2CWriter(conf, c.requests) + c.writer, err = NewP2CWriter(conf, c.requests, sugar) if err != nil { - fmt.Printf("Error creating clickhouse writer: %s\n", err.Error()) + c.logger.Errorf("creating clickhouse writer: %s\n", err.Error()) return c, err } - c.reader, err = NewP2CReader(conf) + c.reader, err = NewP2CReader(conf, sugar) if err != nil { - fmt.Printf("Error creating clickhouse reader: %s\n", err.Error()) + c.logger.Errorf("creating clickhouse reader: %s\n", err.Error()) return c, err } @@ -160,7 +163,7 @@ func (c *p2cServer) process(req remote.WriteRequest) { } func (c *p2cServer) Start() error { - fmt.Println("HTTP server starting...") + c.logger.Info("HTTP server starting...") c.writer.Start() return graceful.RunWithErr(c.conf.HTTPAddr, c.conf.HTTPTimeout, c.mux) } @@ -177,10 +180,10 @@ func (c *p2cServer) Shutdown() { select { case <-wchan: - fmt.Println("Writer shutdown cleanly..") + c.logger.Info("Writer shutdown cleanly..") // All done! case <-time.After(10 * time.Second): - fmt.Println("Writer shutdown timed out, samples will be lost..") + c.logger.Info("Writer shutdown timed out, samples will be lost..") } } diff --git a/writer.go b/writer.go index 1127208..20fae23 100644 --- a/writer.go +++ b/writer.go @@ -9,6 +9,7 @@ import ( "github.com/kshvakov/clickhouse" "github.com/prometheus/client_golang/prometheus" + "go.uber.org/zap" ) var insertSQL = `INSERT INTO %s.%s @@ -24,16 +25,19 @@ type p2cWriter struct { ko prometheus.Counter test prometheus.Counter timings prometheus.Histogram + + logger *zap.SugaredLogger } -func NewP2CWriter(conf *config, reqs chan *p2cRequest) (*p2cWriter, error) { +func NewP2CWriter(conf *config, reqs chan *p2cRequest, sugar *zap.SugaredLogger) (*p2cWriter, error) { var err error w := new(p2cWriter) w.conf = conf w.requests = reqs + w.logger = sugar w.db, err = sql.Open("clickhouse", w.conf.ChDSN) if err != nil { - fmt.Printf("Error connecting to clickhouse: %s\n", err.Error()) + w.logger.With("P2CWriter").Errorf("connecting to clickhouse: %s\n", err.Error()) return w, err } @@ -77,7 +81,7 @@ func (w *p2cWriter) Start() { go func() { w.wg.Add(1) - fmt.Println("Writer starting..") + w.logger.With("P2CWriter").Info("Writer starting..") sql := fmt.Sprintf(insertSQL, w.conf.ChDB, w.conf.ChTable) ok := true for ok { @@ -91,7 +95,7 @@ func (w *p2cWriter) Start() { // get requet and also check if channel is closed req, ok = <-w.requests if !ok { - fmt.Println("Writer stopping..") + w.logger.With("P2CWriter").Info("Writer stopping..") break } reqs = append(reqs, req) @@ -106,7 +110,7 @@ func (w *p2cWriter) Start() { // post them to db all at once tx, err := w.db.Begin() if err != nil { - fmt.Printf("Error: begin transaction: %s\n", err.Error()) + w.logger.With("P2CWriter").Errorf("begin transaction: %s\n", err.Error()) w.ko.Add(1.0) continue } @@ -115,7 +119,7 @@ func (w *p2cWriter) Start() { smt, err := tx.Prepare(sql) for _, req := range reqs { if err != nil { - fmt.Printf("Error: prepare statement: %s\n", err.Error()) + w.logger.With("P2CWriter").Errorf("prepare statement: %s\n", err.Error()) w.ko.Add(1.0) continue } @@ -127,14 +131,14 @@ func (w *p2cWriter) Start() { req.val, req.ts) if err != nil { - fmt.Printf("Error: statement exec: %s\n", err.Error()) + w.logger.With("P2CWriter").Errorf("statement exec: %s\n", err.Error()) w.ko.Add(1.0) } } // commit and record metrics if err = tx.Commit(); err != nil { - fmt.Printf("Error: commit failed: %s\n", err.Error()) + w.logger.With("P2CWriter").Errorf("commit failed: %s\n", err.Error()) w.ko.Add(1.0) } else { w.tx.Add(float64(nmetrics)) @@ -142,7 +146,7 @@ func (w *p2cWriter) Start() { } } - fmt.Println("Writer stopped..") + w.logger.With("P2CWriter").Info("Writer stopped..") w.wg.Done() }() } From 85ebb85b66a97fb5d32c55cb28b99e83d4d4d2f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 16:25:42 +0800 Subject: [PATCH 3/6] fix bug --- reader.go | 20 +++++++++++--------- writer.go | 18 ++++++++++-------- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/reader.go b/reader.go index 5e3113c..130ab6b 100644 --- a/reader.go +++ b/reader.go @@ -12,6 +12,8 @@ import ( "go.uber.org/zap" ) +var readerContent = []interface{}{"component", "reader"} + type p2cReader struct { conf *config db *sql.DB @@ -161,7 +163,7 @@ func NewP2CReader(conf *config, sugar *zap.SugaredLogger) (*p2cReader, error) { r.logger = sugar r.db, err = sql.Open("clickhouse", r.conf.ChDSN) if err != nil { - r.logger.With("P2CReader").Errorf("connecting to clickhouse: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("connecting to clickhouse: %s\n", err.Error()) return r, err } @@ -185,27 +187,27 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) rcount := 0 for _, q := range req.Queries { // remove me.. - r.logger.With("P2CReader").Info("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) + r.logger.With(readerContent...).Info("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) // get the select sql sqlStr, err = r.getSQL(q) - r.logger.With("P2CReader").Info("query: running sql: %s\n\n", sqlStr) + r.logger.With(readerContent...).Info("query: running sql: %s\n\n", sqlStr) if err != nil { - r.logger.With("P2CReader").Errorf("reader: getSQL: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("reader: getSQL: %s\n", err.Error()) return &resp, err } // get the select sql if err != nil { - r.logger.With("P2CReader").Errorf("reader: getSQL: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("reader: getSQL: %s\n", err.Error()) return &resp, err } // todo: metrics on number of errors, rows, selects, timings, etc rows, err = r.db.Query(sqlStr) if err != nil { - r.logger.With("P2CReader").Errorf("query failed: %s", sqlStr) - r.logger.With("P2CReader").Errorf("query error: %s\n", err) + r.logger.With(readerContent...).Errorf("query failed: %s", sqlStr) + r.logger.With(readerContent...).Errorf("query error: %s\n", err) return &resp, err } @@ -221,7 +223,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) value float64 ) if err = rows.Scan(&cnt, &t, &name, &tags, &value); err != nil { - r.logger.With("P2CReader").Errorf("scan: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("scan: %s\n", err.Error()) } // remove this.. //fmt.Printf(fmt.Sprintf("%d,%d,%s,%s,%f\n", cnt, t, name, strings.Join(tags, ":"), value)) @@ -247,7 +249,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) resp.Results[0].Timeseries = append(resp.Results[0].Timeseries, ts) } - r.logger.With("P2CReader").Info("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) + r.logger.With(readerContent...).Info("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) return &resp, nil diff --git a/writer.go b/writer.go index 20fae23..d17a70e 100644 --- a/writer.go +++ b/writer.go @@ -16,6 +16,8 @@ var insertSQL = `INSERT INTO %s.%s (date, name, tags, val, ts) VALUES (?, ?, ?, ?, ?)` +var writerContent = []interface{}{"component", "writer"} + type p2cWriter struct { conf *config requests chan *p2cRequest @@ -37,7 +39,7 @@ func NewP2CWriter(conf *config, reqs chan *p2cRequest, sugar *zap.SugaredLogger) w.logger = sugar w.db, err = sql.Open("clickhouse", w.conf.ChDSN) if err != nil { - w.logger.With("P2CWriter").Errorf("connecting to clickhouse: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("connecting to clickhouse: %s\n", err.Error()) return w, err } @@ -81,7 +83,7 @@ func (w *p2cWriter) Start() { go func() { w.wg.Add(1) - w.logger.With("P2CWriter").Info("Writer starting..") + w.logger.With(writerContent...).Info("Writer starting..") sql := fmt.Sprintf(insertSQL, w.conf.ChDB, w.conf.ChTable) ok := true for ok { @@ -95,7 +97,7 @@ func (w *p2cWriter) Start() { // get requet and also check if channel is closed req, ok = <-w.requests if !ok { - w.logger.With("P2CWriter").Info("Writer stopping..") + w.logger.With(writerContent...).Info("Writer stopping..") break } reqs = append(reqs, req) @@ -110,7 +112,7 @@ func (w *p2cWriter) Start() { // post them to db all at once tx, err := w.db.Begin() if err != nil { - w.logger.With("P2CWriter").Errorf("begin transaction: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("begin transaction: %s\n", err.Error()) w.ko.Add(1.0) continue } @@ -119,7 +121,7 @@ func (w *p2cWriter) Start() { smt, err := tx.Prepare(sql) for _, req := range reqs { if err != nil { - w.logger.With("P2CWriter").Errorf("prepare statement: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("prepare statement: %s\n", err.Error()) w.ko.Add(1.0) continue } @@ -131,14 +133,14 @@ func (w *p2cWriter) Start() { req.val, req.ts) if err != nil { - w.logger.With("P2CWriter").Errorf("statement exec: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("statement exec: %s\n", err.Error()) w.ko.Add(1.0) } } // commit and record metrics if err = tx.Commit(); err != nil { - w.logger.With("P2CWriter").Errorf("commit failed: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("commit failed: %s\n", err.Error()) w.ko.Add(1.0) } else { w.tx.Add(float64(nmetrics)) @@ -146,7 +148,7 @@ func (w *p2cWriter) Start() { } } - w.logger.With("P2CWriter").Info("Writer stopped..") + w.logger.With(writerContent...).Info("Writer stopped..") w.wg.Done() }() } From 482fc5a1552e07e56b18dae82cd48886a1c522dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 16:45:02 +0800 Subject: [PATCH 4/6] add debug mode --- main.go | 12 +++++++++++- reader.go | 6 +++--- writer.go | 6 +++--- 3 files changed, 17 insertions(+), 7 deletions(-) diff --git a/main.go b/main.go index 4c4cbb4..c693e50 100644 --- a/main.go +++ b/main.go @@ -30,6 +30,7 @@ type config struct { var ( versionFlag bool + debug bool ) func main() { @@ -44,7 +45,13 @@ func main() { return } - logger, _ := zap.NewProduction() + var logger *zap.Logger + if debug { + logger, _ = zap.NewDevelopment() + } else { + logger, _ = zap.NewProduction() + } + defer logger.Sync() // flushes buffer, if any sugar := logger.Sugar() @@ -69,6 +76,9 @@ func parseFlags() *config { // print version? flag.BoolVar(&versionFlag, "version", false, "Version") + // turn on debug? + flag.BoolVar(&debug, "debug", false, "turn on debug mode") + // clickhouse dsn ddsn := "tcp://127.0.0.1:9000?username=&password=&database=metrics&" + "read_timeout=10&write_timeout=10&alt_hosts=" diff --git a/reader.go b/reader.go index 130ab6b..67b2344 100644 --- a/reader.go +++ b/reader.go @@ -187,11 +187,11 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) rcount := 0 for _, q := range req.Queries { // remove me.. - r.logger.With(readerContent...).Info("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) + r.logger.With(readerContent...).Debug("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) // get the select sql sqlStr, err = r.getSQL(q) - r.logger.With(readerContent...).Info("query: running sql: %s\n\n", sqlStr) + r.logger.With(readerContent...).Debug("query: running sql: %s\n\n", sqlStr) if err != nil { r.logger.With(readerContent...).Errorf("reader: getSQL: %s\n", err.Error()) return &resp, err @@ -249,7 +249,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) resp.Results[0].Timeseries = append(resp.Results[0].Timeseries, ts) } - r.logger.With(readerContent...).Info("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) + r.logger.With(readerContent...).Debug("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) return &resp, nil diff --git a/writer.go b/writer.go index d17a70e..32b0b74 100644 --- a/writer.go +++ b/writer.go @@ -83,7 +83,7 @@ func (w *p2cWriter) Start() { go func() { w.wg.Add(1) - w.logger.With(writerContent...).Info("Writer starting..") + w.logger.With(writerContent...).Debug("Writer starting..") sql := fmt.Sprintf(insertSQL, w.conf.ChDB, w.conf.ChTable) ok := true for ok { @@ -97,7 +97,7 @@ func (w *p2cWriter) Start() { // get requet and also check if channel is closed req, ok = <-w.requests if !ok { - w.logger.With(writerContent...).Info("Writer stopping..") + w.logger.With(writerContent...).Debug("Writer stopping..") break } reqs = append(reqs, req) @@ -148,7 +148,7 @@ func (w *p2cWriter) Start() { } } - w.logger.With(writerContent...).Info("Writer stopped..") + w.logger.With(writerContent...).Debug("Writer stopped..") w.wg.Done() }() } From 4539b2d8aa7174715a66dcc2eca297d71969da8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 16:48:41 +0800 Subject: [PATCH 5/6] update glide --- glide.lock | 16 ++++++++++++++-- glide.yaml | 2 ++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/glide.lock b/glide.lock index e17fba9..f1b072a 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 2936591cb6ef55dc5517f96d731851b2135281bd2200919933219089b837e517 -updated: 2017-06-11T21:38:09.459666062+10:00 +hash: 9205cd1cc00a2ac50c6d793dafb1850c244853cabb3cfab85be7b6f524dd2464 +updated: 2018-07-12T16:47:57.80074341+08:00 imports: - name: github.com/beorn7/perks version: 4c0e84591b9aa9e6dcfdf3e020114cd81f89d5f9 @@ -72,6 +72,18 @@ imports: - leveldb/storage - leveldb/table - leveldb/util +- name: go.uber.org/atomic + version: 1ea20fb1cbb1cc08cbd0d913a96dead89aa18289 +- name: go.uber.org/multierr + version: 3c4937480c32f4c13a875a1829af76c98ca3d40a +- name: go.uber.org/zap + version: eeedf312bc6c57391d84767a4cd413f02a917974 + subpackages: + - buffer + - internal/bufferpool + - internal/color + - internal/exit + - zapcore - name: golang.org/x/net version: 1a68b1313cf4ad7778376e82641197b60c02f65c subpackages: diff --git a/glide.yaml b/glide.yaml index 4d790d8..d10bb39 100644 --- a/glide.yaml +++ b/glide.yaml @@ -18,3 +18,5 @@ import: - package: gopkg.in/tylerb/graceful.v1 version: v1.2.15 - package: github.com/kshvakov/clickhouse +- package: go.uber.org/zap + version: ^1.8.0 From 4aa1b200051bb3a4468a95ea1dd11b9fcd73da62 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E4=BC=9A=E5=86=9B?= Date: Thu, 12 Jul 2018 20:20:26 +0800 Subject: [PATCH 6/6] add ping --- reader.go | 26 ++++++++++++++++++-------- writer.go | 25 +++++++++++++++++-------- 2 files changed, 35 insertions(+), 16 deletions(-) diff --git a/reader.go b/reader.go index 67b2344..2b0fa14 100644 --- a/reader.go +++ b/reader.go @@ -7,6 +7,7 @@ import ( "fmt" "strings" + "github.com/kshvakov/clickhouse" "github.com/prometheus/common/model" "github.com/prometheus/prometheus/storage/remote" "go.uber.org/zap" @@ -163,7 +164,16 @@ func NewP2CReader(conf *config, sugar *zap.SugaredLogger) (*p2cReader, error) { r.logger = sugar r.db, err = sql.Open("clickhouse", r.conf.ChDSN) if err != nil { - r.logger.With(readerContent...).Errorf("connecting to clickhouse: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("connecting to clickhouse: %s", err.Error()) + return r, err + } + + if err := r.db.Ping(); err != nil { + if exception, ok := err.(*clickhouse.Exception); ok { + r.logger.With(readerContent...).Errorf("[%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace) + } else { + r.logger.With(readerContent...).Error(err.Error()) + } return r, err } @@ -187,19 +197,19 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) rcount := 0 for _, q := range req.Queries { // remove me.. - r.logger.With(readerContent...).Debug("\nquery: start: %d, end: %d\n\n", q.StartTimestampMs, q.EndTimestampMs) + r.logger.With(readerContent...).Debug("\nquery: start: %d, end: %d", q.StartTimestampMs, q.EndTimestampMs) // get the select sql sqlStr, err = r.getSQL(q) - r.logger.With(readerContent...).Debug("query: running sql: %s\n\n", sqlStr) + r.logger.With(readerContent...).Debug("query: running sql: %s", sqlStr) if err != nil { - r.logger.With(readerContent...).Errorf("reader: getSQL: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("reader: getSQL: %s", err.Error()) return &resp, err } // get the select sql if err != nil { - r.logger.With(readerContent...).Errorf("reader: getSQL: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("reader: getSQL: %s", err.Error()) return &resp, err } @@ -207,7 +217,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) rows, err = r.db.Query(sqlStr) if err != nil { r.logger.With(readerContent...).Errorf("query failed: %s", sqlStr) - r.logger.With(readerContent...).Errorf("query error: %s\n", err) + r.logger.With(readerContent...).Errorf("query error: %s", err) return &resp, err } @@ -223,7 +233,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) value float64 ) if err = rows.Scan(&cnt, &t, &name, &tags, &value); err != nil { - r.logger.With(readerContent...).Errorf("scan: %s\n", err.Error()) + r.logger.With(readerContent...).Errorf("scan: %s", err.Error()) } // remove this.. //fmt.Printf(fmt.Sprintf("%d,%d,%s,%s,%f\n", cnt, t, name, strings.Join(tags, ":"), value)) @@ -249,7 +259,7 @@ func (r *p2cReader) Read(req *remote.ReadRequest) (*remote.ReadResponse, error) resp.Results[0].Timeseries = append(resp.Results[0].Timeseries, ts) } - r.logger.With(readerContent...).Debug("query: returning %d rows for %d queries\n", rcount, len(req.Queries)) + r.logger.With(readerContent...).Debug("query: returning %d rows for %d queries", rcount, len(req.Queries)) return &resp, nil diff --git a/writer.go b/writer.go index 32b0b74..b6cd48b 100644 --- a/writer.go +++ b/writer.go @@ -39,7 +39,16 @@ func NewP2CWriter(conf *config, reqs chan *p2cRequest, sugar *zap.SugaredLogger) w.logger = sugar w.db, err = sql.Open("clickhouse", w.conf.ChDSN) if err != nil { - w.logger.With(writerContent...).Errorf("connecting to clickhouse: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("connecting to clickhouse: %s", err.Error()) + return w, err + } + + if err := w.db.Ping(); err != nil { + if exception, ok := err.(*clickhouse.Exception); ok { + w.logger.With(writerContent...).Errorf("[%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace) + } else { + w.logger.With(writerContent...).Error(err.Error()) + } return w, err } @@ -83,7 +92,7 @@ func (w *p2cWriter) Start() { go func() { w.wg.Add(1) - w.logger.With(writerContent...).Debug("Writer starting..") + w.logger.With(writerContent...).Info("Writer starting..") sql := fmt.Sprintf(insertSQL, w.conf.ChDB, w.conf.ChTable) ok := true for ok { @@ -97,7 +106,7 @@ func (w *p2cWriter) Start() { // get requet and also check if channel is closed req, ok = <-w.requests if !ok { - w.logger.With(writerContent...).Debug("Writer stopping..") + w.logger.With(writerContent...).Info("Writer stopping..") break } reqs = append(reqs, req) @@ -112,7 +121,7 @@ func (w *p2cWriter) Start() { // post them to db all at once tx, err := w.db.Begin() if err != nil { - w.logger.With(writerContent...).Errorf("begin transaction: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("begin transaction: %s", err.Error()) w.ko.Add(1.0) continue } @@ -121,7 +130,7 @@ func (w *p2cWriter) Start() { smt, err := tx.Prepare(sql) for _, req := range reqs { if err != nil { - w.logger.With(writerContent...).Errorf("prepare statement: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("prepare statement: %s", err.Error()) w.ko.Add(1.0) continue } @@ -133,14 +142,14 @@ func (w *p2cWriter) Start() { req.val, req.ts) if err != nil { - w.logger.With(writerContent...).Errorf("statement exec: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("statement exec: %s", err.Error()) w.ko.Add(1.0) } } // commit and record metrics if err = tx.Commit(); err != nil { - w.logger.With(writerContent...).Errorf("commit failed: %s\n", err.Error()) + w.logger.With(writerContent...).Errorf("commit failed: %s", err.Error()) w.ko.Add(1.0) } else { w.tx.Add(float64(nmetrics)) @@ -148,7 +157,7 @@ func (w *p2cWriter) Start() { } } - w.logger.With(writerContent...).Debug("Writer stopped..") + w.logger.With(writerContent...).Info("Writer stopped..") w.wg.Done() }() }