From a6c6f4d1c1d186df3c82181d826d95d79b8c819c Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 13:36:58 -0500 Subject: [PATCH 01/19] changed some json tag names to Lower Camel Case to be consistent with the rest --- cluster.go | 2 +- ctl/import.go | 2 +- handler.go | 6 +++--- handler_test.go | 10 +++++----- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/cluster.go b/cluster.go index 6290f0050..f92b8c6e9 100644 --- a/cluster.go +++ b/cluster.go @@ -20,7 +20,7 @@ const ( // Node represents a node in the cluster. type Node struct { Host string `json:"host"` - InternalHost string `json:"internal_host"` + InternalHost string `json:"internalHost"` } // Nodes represents a list of nodes. diff --git a/ctl/import.go b/ctl/import.go index ad3281813..df8fb00eb 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -27,7 +27,7 @@ type ImportCommand struct { Paths []string `json:"paths"` // Size of buffer used to chunk import. - BufferSize int `json:"buffer-size"` + BufferSize int `json:"bufferSize"` // Reusable client. Client *pilosa.Client `json:"-"` diff --git a/handler.go b/handler.go index c9140867b..a6309e44a 100644 --- a/handler.go +++ b/handler.go @@ -214,7 +214,7 @@ func (h *Handler) handleGetSliceMax(w http.ResponseWriter, r *http.Request) { } type sliceMaxResponse struct { - MaxSlices map[string]uint64 `json:"MaxSlices"` + MaxSlices map[string]uint64 `json:"maxSlices"` } // handleGetDBs handles GET /db request. @@ -391,7 +391,7 @@ func (h *Handler) handlePatchDBTimeQuantum(w http.ResponseWriter, r *http.Reques } type patchDBTimeQuantumRequest struct { - TimeQuantum string `json:"time_quantum"` + TimeQuantum string `json:"timeQuantum"` } type patchDBTimeQuantumResponse struct{} @@ -611,7 +611,7 @@ func (h *Handler) handlePatchFrameTimeQuantum(w http.ResponseWriter, r *http.Req } type patchFrameTimeQuantumRequest struct { - TimeQuantum string `json:"time_quantum"` + TimeQuantum string `json:"timeQuantum"` } type patchFrameTimeQuantumResponse struct{} diff --git a/handler_test.go b/handler_test.go index 2611da50e..478720e23 100644 --- a/handler_test.go +++ b/handler_test.go @@ -83,7 +83,7 @@ func TestHandler_MaxSlices(t *testing.T) { h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max", nil)) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" { + } else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -123,7 +123,7 @@ func TestHandler_MaxSlices_Inverse(t *testing.T) { h.ServeHTTP(w, MustNewHTTPRequest("GET", "/slices/max?inverse=true", nil)) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if body := w.Body.String(); body != `{"MaxSlices":{"d0":3,"d1":0}}`+"\n" { + } else if body := w.Body.String(); body != `{"maxSlices":{"d0":3,"d1":0}}`+"\n" { t.Fatalf("unexpected body: %s", body) } } @@ -556,7 +556,7 @@ func TestHandler_SetDBTimeQuantum(t *testing.T) { h := NewHandler() h.Index = idx.Index w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`))) + h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{}`+"\n" { @@ -579,7 +579,7 @@ func TestHandler_SetFrameTimeQuantum(t *testing.T) { h := NewHandler() h.Index = idx.Index w := httptest.NewRecorder() - h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"time_quantum":"ymdh"}`))) + h.ServeHTTP(w, MustNewHTTPRequest("PATCH", "/db/d0/frame/f1/time-quantum", strings.NewReader(`{"timeQuantum":"ymdh"}`))) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) } else if body := w.Body.String(); body != `{}`+"\n" { @@ -763,7 +763,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) { h.ServeHTTP(w, r) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if w.Body.String() != `[{"host":"host1","internal_host":""},{"host":"host2","internal_host":""}]`+"\n" { + } else if w.Body.String() != `[{"host":"host1","internalHost":""},{"host":"host2","internalHost":""}]`+"\n" { t.Fatalf("unexpected body: %q", w.Body.String()) } } From 68d9d1543ff1a27a83dd92f04cd3f9fc049b8cc2 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 16:07:30 -0500 Subject: [PATCH 02/19] rename messenger_test to align with broadcast --- messenger_test.go => broadcast_test.go | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename messenger_test.go => broadcast_test.go (100%) diff --git a/messenger_test.go b/broadcast_test.go similarity index 100% rename from messenger_test.go rename to broadcast_test.go From 8440bc8b3ca90d6b4742037bce3a6fc75119099e Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 16:27:39 -0500 Subject: [PATCH 03/19] add cache stats --- cache.go | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/cache.go b/cache.go index 7da28eb74..597d65732 100644 --- a/cache.go +++ b/cache.go @@ -35,12 +35,16 @@ type Cache interface { // Returns an ordered list of the top ranked bitmaps. Top() []BitmapPair + + // SetStats defines the stats client used in the cache + SetStats(s StatsClient) } // LRUCache represents a least recently used Cache implemenation. type LRUCache struct { cache *lru.Cache counts map[uint64]uint64 + stats StatsClient } // NewLRUCache returns a new instance of LRUCache. @@ -48,6 +52,7 @@ func NewLRUCache(maxEntries uint32) *LRUCache { c := &LRUCache{ cache: lru.New(int(maxEntries)), counts: make(map[uint64]uint64), + stats: NopStatsClient, } c.cache.OnEvicted = c.onEvicted return c @@ -74,10 +79,14 @@ func (c *LRUCache) Get(bitmapID uint64) uint64 { func (c *LRUCache) Len() int { return c.cache.Len() } // Invalidate is a no-op. -func (c *LRUCache) Invalidate() {} +func (c *LRUCache) Invalidate() { + c.stats.Gauge("LRUCache", float64(c.cache.Len())) +} // Recalculate is a no-op. -func (c *LRUCache) Recalculate() {} +func (c *LRUCache) Recalculate() { + c.stats.Gauge("LRUCache", float64(c.cache.Len())) +} // BitmapIDs returns a list of all bitmap IDs in the cache. func (c *LRUCache) BitmapIDs() []uint64 { @@ -102,6 +111,11 @@ func (c *LRUCache) Top() []BitmapPair { return a } +// SetStats passes the stats client used the the frame that owns the cache +func (c *LRUCache) SetStats(s StatsClient) { + c.stats = s +} + func (c *LRUCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) } // Ensure LRUCache implements Cache. @@ -125,6 +139,8 @@ type RankCache struct { // thresholdValue is the value of the last item in the cache thresholdValue uint64 + + stats StatsClient } // NewRankCache returns a new instance of RankCache. @@ -133,6 +149,7 @@ func NewRankCache(maxEntries uint32) *RankCache { maxEntries: maxEntries, thresholdBuffer: int(ThresholdFactor * float64(maxEntries)), entries: make(map[uint64]uint64), + stats: NopStatsClient, } } @@ -222,7 +239,10 @@ func (c *RankCache) recalculate() { // Store the count of the item at the threshold index. c.rankings = rankings - if len(c.rankings) > int(c.maxEntries) { + length := len(c.rankings) + c.stats.Gauge("RankCache", float64(length)) + + if length > int(c.maxEntries) { c.thresholdValue = rankings[c.maxEntries].Count c.rankings = c.rankings[0:c.maxEntries] } else { @@ -242,6 +262,11 @@ func (c *RankCache) recalculate() { } } +// SetStats passes the stats client used the the frame that owns the cache +func (c *RankCache) SetStats(s StatsClient) { + c.stats = s +} + // Top returns an ordered list of bitmaps. func (c *RankCache) Top() []BitmapPair { return c.rankings } From 20bc2396f8f925f3faa53ce2bd8e19624914720e Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 16:51:55 -0500 Subject: [PATCH 04/19] add stats metric options to config --- cmd/server.go | 4 +++- config.go | 8 ++++++++ ctl/config.go | 4 ++++ 3 files changed, 15 insertions(+), 1 deletion(-) diff --git a/cmd/server.go b/cmd/server.go index 358793af3..4e5d96114 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -87,7 +87,9 @@ on the configured port.`, flags.StringVarP(&Server.Config.Cluster.Type, "cluster.type", "", "static", "Determine how the cluster handles membership and state sharing. Choose from [static, http, gossip]") flags.StringVarP(&Server.Config.Cluster.GossipSeed, "cluster.gossip-seed", "", "", "Host with which to seed the gossip membership.") flags.StringVarP(&Server.Config.Cluster.InternalPort, "cluster.internal-port", "", "", "Port to which pilosa should bind for internal state sharing.") - + flags.StringVarP(&Server.Config.Metric.Service, "metric.service", "", "noop", "Default URI on which pilosa should listen.") + flags.StringVarP(&Server.Config.Metric.Host, "metric.host", "", "", "Default URI to send metrics.") + flags.DurationVarP((*time.Duration)(&Server.Config.Metric.PollingInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") return serveCmd } diff --git a/config.go b/config.go index d5d725ece..a0b51a503 100644 --- a/config.go +++ b/config.go @@ -8,6 +8,7 @@ const ( DefaultPort = "10101" DefaultClusterType = "static" DefaultInternalPort = "14000" + DefaultMetrics = "noop" ) // Config represents the configuration for the command. @@ -34,6 +35,12 @@ type Config struct { } `toml:"anti-entropy"` LogPath string `toml:"log-path"` + + Metric struct { + Service string `toml:"service"` + Host string `toml:"host"` + PollingInterval Duration `toml:"interval"` + } `toml:"metrics"` } // NewConfig returns an instance of Config with default options. @@ -47,6 +54,7 @@ func NewConfig() *Config { c.Cluster.Hosts = []string{} c.Cluster.InternalHosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) + c.Metric.Service = DefaultMetrics return c } diff --git a/ctl/config.go b/ctl/config.go index dbac574e7..3d940aff1 100644 --- a/ctl/config.go +++ b/ctl/config.go @@ -37,6 +37,10 @@ bind = "localhost:10101" [anti-entropy] interval = "10m0s" +[metrics] + service = "statsd" + host = "127.0.0.1:8125" + [profile] cpu = "" cpu-time = "30s" From df2e282bdfea2c729cf90f21ade047b969496ac9 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 16:52:49 -0500 Subject: [PATCH 05/19] fix server tests to bind to new node if changing the node set --- cmd/server_test.go | 75 ++++++++++++++++++++++++---------------------- 1 file changed, 40 insertions(+), 35 deletions(-) diff --git a/cmd/server_test.go b/cmd/server_test.go index e51825e60..edd61c601 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -29,25 +29,25 @@ func TestServerConfig(t *testing.T) { tests := []commandTest{ // TEST 0 { - args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "example.com:10101,example.com:10110"}, + args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "example.com:10111,example.com:10110", "--bind", "example.com:10111"}, env: map[string]string{"PILOSA_DATA_DIR": "/tmp/myEnvDatadir", "PILOSA_CLUSTER.POLL_INTERVAL": "3m2s"}, cfgFileContent: ` -data-dir = "/tmp/myFileDatadir" -bind = "localhost:0" + data-dir = "/tmp/myFileDatadir" + bind = "localhost:0" -[cluster] - poll-interval = "45s" - replicas = 2 - hosts = [ - "localhost:19444", - ] -`, + [cluster] + poll-interval = "45s" + replicas = 2 + hosts = [ + "localhost:19444", + ] + `, validation: func() error { v := validator{} v.Check(cmd.Server.Config.DataDir, actualDataDir) - v.Check(cmd.Server.Config.Host, "localhost:0") + v.Check(cmd.Server.Config.Host, "example.com:10111") v.Check(cmd.Server.Config.Cluster.ReplicaN, 2) - v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10101", "example.com:10110"}) + v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10111", "example.com:10110"}) v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Second*182)) return v.Error() }, @@ -55,17 +55,17 @@ bind = "localhost:0" // TEST 1 { args: []string{"server", "--anti-entropy.interval", "9m0s"}, - env: map[string]string{"PILOSA_CLUSTER.HOSTS": "example.com:1110,example.com:1111"}, + env: map[string]string{"PILOSA_CLUSTER.HOSTS": "example.com:1110,example.com:1111", "PILOSA_BIND": "example.com:1110"}, cfgFileContent: ` -bind = "localhost:0" -data-dir = "` + actualDataDir + `" -[cluster] - hosts = [ - "localhost:19444", - ] -[plugins] - path = "/var/sloth" -`, + bind = "localhost:0" + data-dir = "` + actualDataDir + `" + [cluster] + hosts = [ + "localhost:19444", + ] + [plugins] + path = "/var/sloth" + `, validation: func() error { v := validator{} v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:1110", "example.com:1111"}) @@ -79,19 +79,22 @@ data-dir = "` + actualDataDir + `" args: []string{"server", "--log-path", logFile.Name()}, env: map[string]string{"PILOSA_PROFILE.CPU_TIME": "1m"}, cfgFileContent: ` -bind = "localhost:0" -data-dir = "` + actualDataDir + `" -[cluster] - poll-interval = "2m0s" - hosts = [ - "localhost:19444", - ] -[anti-entropy] - interval = "11m0s" -[profile] - cpu = "` + profFile.Name() + `" - cpu-time = "35s" -`, + bind = "localhost:19444" + data-dir = "` + actualDataDir + `" + [cluster] + poll-interval = "2m0s" + hosts = [ + "localhost:19444", + ] + [anti-entropy] + interval = "11m0s" + [profile] + cpu = "` + profFile.Name() + `" + cpu-time = "35s" + [metric] + service = "statsd" + host = "127.0.0.1:8125" + `, validation: func() error { v := validator{} v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"}) @@ -100,6 +103,8 @@ data-dir = "` + actualDataDir + `" v.Check(cmd.Server.CPUProfile, profFile.Name()) v.Check(cmd.Server.CPUTime, time.Minute) v.Check(cmd.Server.Config.LogPath, logFile.Name()) + v.Check(cmd.Server.Config.Metric.Service, "statsd") + v.Check(cmd.Server.Config.Metric.Host, "127.0.0.1:8125") if v.Error() != nil { return v.Error() } From f3189889d422777e056c9ee56dba391413bc1331 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 16:59:56 -0500 Subject: [PATCH 06/19] add support for CreateDB and CreateFrame without POST data (i.e. use defaults) --- handler.go | 33 +++++++++++++++++++++++++++------ 1 file changed, 27 insertions(+), 6 deletions(-) diff --git a/handler.go b/handler.go index a6309e44a..b94a9a0ed 100644 --- a/handler.go +++ b/handler.go @@ -309,6 +309,15 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) { return } + // Send the delete message to all nodes. + err := h.Broadcaster.SendSync( + &internal.DeleteDBMessage{ + DB: dbName, + }) + if err != nil { + h.logger().Printf("problem sending DeleteDB message: %s", err) + } + // Encode response. if err := json.NewEncoder(w).Encode(deleteDBResponse{}); err != nil { h.logger().Printf("response encoding error: %s", err) @@ -323,13 +332,17 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Decode request. var req postDBRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + err := json.NewDecoder(r.Body).Decode(&req) + if err == io.EOF { + // If no data was provided (EOF), we still create the database + // with default values. + } else if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } // Create database. - _, err := h.Index.CreateDB(dbName, req.Options) + _, err = h.Index.CreateDB(dbName, req.Options) if err == ErrDatabaseExists { http.Error(w, err.Error(), http.StatusConflict) return @@ -340,11 +353,15 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Send the delete message to all nodes. err = h.Broadcaster.SendSync( - &internal.DeleteDBMessage{ + &internal.CreateDBMessage{ DB: dbName, + Meta: &internal.DBMeta{ + ColumnLabel: req.Options.ColumnLabel, + TimeQuantum: string(req.Options.TimeQuantum), + }, }) if err != nil { - h.logger().Printf("problem sending DeleteDB message: %s", err) + h.logger().Printf("problem sending CreateDB message: %s", err) } // Encode response. @@ -460,7 +477,11 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { // Decode request. var req postFrameRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + err := json.NewDecoder(r.Body).Decode(&req) + if err == io.EOF { + // If no data was provided (EOF), we still create the frame + // with default values. + } else if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -473,7 +494,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { } // Create frame. - _, err := db.CreateFrame(frameName, req.Options) + _, err = db.CreateFrame(frameName, req.Options) if err == ErrFrameExists { http.Error(w, err.Error(), http.StatusConflict) return From c101f647c819fb76450769376dd7c5632ca19ad0 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 17:08:10 -0500 Subject: [PATCH 07/19] adjust some handler comments. remove debugging line --- handler.go | 8 ++++---- server.go | 1 - 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/handler.go b/handler.go index b94a9a0ed..d9c79eeba 100644 --- a/handler.go +++ b/handler.go @@ -309,7 +309,7 @@ func (h *Handler) handleDeleteDB(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the delete database message to all nodes. err := h.Broadcaster.SendSync( &internal.DeleteDBMessage{ DB: dbName, @@ -351,7 +351,7 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the create database message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateDBMessage{ DB: dbName, @@ -503,7 +503,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { return } - // Send the create message to all nodes. + // Send the create frame message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateFrameMessage{ DB: dbName, @@ -575,7 +575,7 @@ func (h *Handler) handleDeleteFrame(w http.ResponseWriter, r *http.Request) { return } - // Send the delete message to all nodes. + // Send the delete frame message to all nodes. err := h.Broadcaster.SendSync( &internal.DeleteFrameMessage{ DB: dbName, diff --git a/server.go b/server.go index 229214805..e93b8e111 100644 --- a/server.go +++ b/server.go @@ -265,7 +265,6 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return err } case *internal.DeleteDBMessage: - fmt.Println("DELETE:", obj.DB) if err := s.Index.DeleteDB(obj.DB); err != nil { return err } From 8fe0ee47cc854248a4d4d0e35e9f73b872d68dd5 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 17:21:08 -0500 Subject: [PATCH 08/19] specify the StatsD host string --- datadog/datadog.go | 7 +++++-- datadog/datadog_test.go | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/datadog/datadog.go b/datadog/datadog.go index ff5e728b5..27ef58644 100644 --- a/datadog/datadog.go +++ b/datadog/datadog.go @@ -10,6 +10,9 @@ import ( "github.com/pilosa/pilosa" ) +// StatsD protocal wrapper using the DataDog library that added Tags to the StatsD protocal +// statsD defailt host is "127.0.0.1:8125" + const ( // Rate represents a metric rate of 1/sec. Rate = 1 @@ -30,8 +33,8 @@ type StatsClient struct { } // NewStatsClient returns a new instance of StatsClient. -func NewStatsClient() (*StatsClient, error) { - c, err := statsd.NewBuffered("127.0.0.1:8125", BufferLen) +func NewStatsClient(host string) (*StatsClient, error) { + c, err := statsd.NewBuffered(host, BufferLen) if err != nil { return nil, err } diff --git a/datadog/datadog_test.go b/datadog/datadog_test.go index 00c4bf69a..b5a8bf5ee 100644 --- a/datadog/datadog_test.go +++ b/datadog/datadog_test.go @@ -9,7 +9,7 @@ import ( func TestStatsClient_WithTags(t *testing.T) { // Create a new client. - c, err := datadog.NewStatsClient() + c, err := datadog.NewStatsClient("localhost:19444") if err != nil { t.Fatal(err) } From ef5fbddf37e6d62d96373b931999a601b09d15b5 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 17:22:00 -0500 Subject: [PATCH 09/19] The server uses the stats service specified in the config --- server/server.go | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/server/server.go b/server/server.go index ec2b55c7c..7e4171d3d 100644 --- a/server/server.go +++ b/server/server.go @@ -17,6 +17,7 @@ import ( "time" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/datadog" "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/httpbroadcast" ) @@ -90,6 +91,7 @@ func (m *Command) Run(args ...string) (err error) { } func (m *Command) SetupServer() error { + var err error cluster := pilosa.NewCluster() cluster.ReplicaN = m.Config.Cluster.ReplicaN @@ -118,9 +120,12 @@ func (m *Command) SetupServer() error { // Configure index. fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir) m.Server.Index.Path = m.Config.DataDir - m.Server.Index.Stats = pilosa.NewExpvarStatsClient() + m.Server.MetricInterval = time.Duration(m.Config.Metric.PollingInterval) + m.Server.Index.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host) + if err != nil { + return err + } - var err error m.Server.Host, err = normalizeHost(m.Config.Host) if err != nil { return err @@ -201,3 +206,15 @@ func (m *Command) Close() error { } return serveErr } + +// NewStatsClient creates a stats client from the config +func NewStatsClient(name string, host string) (pilosa.StatsClient, error) { + switch name { + case "expvar": + return pilosa.NewExpvarStatsClient(), nil + case "statsd": + return datadog.NewStatsClient(host) + default: + return pilosa.NopStatsClient, nil + } +} From 7da469d62df730f2c95a7dc01279e2e99e510dff Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Wed, 19 Apr 2017 17:22:40 -0500 Subject: [PATCH 10/19] measure runtime stats added package to measure when garbage collection occurs --- glide.lock | 12 +++++++----- glide.yaml | 1 + server.go | 41 ++++++++++++++++++++++++++++++++++++++++- 3 files changed, 48 insertions(+), 6 deletions(-) diff --git a/glide.lock b/glide.lock index c142737ff..7ba5cdab6 100644 --- a/glide.lock +++ b/glide.lock @@ -1,5 +1,5 @@ -hash: 4bdea17c62dcd469584382515052e7a246fae6deefc2d62da70bf768e18e1f0c -updated: 2017-04-19T10:51:10.409094081-05:00 +hash: a6889bf5334164aa4aebd670eee45444f1bdfa1888642d290acf33e18e57f293 +updated: 2017-04-19T17:19:43.584486333-05:00 imports: - name: github.com/armon/go-metrics version: 97c69685293dce4c0a2d0b19535179bbc976e4d2 @@ -7,6 +7,8 @@ imports: version: 4b1ebc1869ad66568b313d0dc410e2be72670dda - name: github.com/BurntSushi/toml version: 99064174e013895bbd9b025c31100bd1d9b590ca +- name: github.com/CAFxX/gcnotifier + version: adea3e70515666981da25214a7d3e377e4841c22 - name: github.com/DataDog/datadog-go version: 909c02b65dd8a52e8fa6072db9752a112227cf21 subpackages: @@ -46,10 +48,10 @@ imports: subpackages: - hcl/ast - hcl/parser - - hcl/scanner - - hcl/strconv - hcl/token - json/parser + - hcl/scanner + - hcl/strconv - json/scanner - json/token - name: github.com/hashicorp/memberlist @@ -101,4 +103,4 @@ imports: - unicode/norm - name: gopkg.in/yaml.v2 version: a3f3340b5840cee44f372bddb5880fcbc419b46a -testImports: [] +devImports: [] diff --git a/glide.yaml b/glide.yaml index a7c0e98eb..4b1ff6038 100644 --- a/glide.yaml +++ b/glide.yaml @@ -33,3 +33,4 @@ import: version: ^1.3.0 - package: github.com/hashicorp/memberlist - package: golang.org/x/sync +- package: github.com/CAFxX/gcnotifier diff --git a/server.go b/server.go index 229214805..68337049d 100644 --- a/server.go +++ b/server.go @@ -10,10 +10,12 @@ import ( "net/http" "net/url" "os" + "runtime" "strconv" "sync" "time" + "github.com/CAFxX/gcnotifier" "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" ) @@ -46,6 +48,7 @@ type Server struct { // Background monitoring intervals. AntiEntropyInterval time.Duration PollingInterval time.Duration + MetricInterval time.Duration LogOutput io.Writer } @@ -62,6 +65,7 @@ func NewServer() *Server { AntiEntropyInterval: DefaultAntiEntropyInterval, PollingInterval: DefaultPollingInterval, + MetricInterval: 0, LogOutput: os.Stderr, } @@ -131,9 +135,10 @@ func (s *Server) Open() error { go func() { http.Serve(ln, s.Handler) }() // Start background monitoring. - s.wg.Add(2) + s.wg.Add(3) go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() go func() { defer s.wg.Done(); s.monitorMaxSlices() }() + go func() { defer s.wg.Done(); s.monitorRuntime() }() return nil } @@ -363,3 +368,37 @@ func checkMaxSlices(hostport string) (map[string]uint64, error) { return pb.MaxSlices, nil } + +// monitorRuntime periodically polls the Go runtime metrics. +func (s *Server) monitorRuntime() { + if s.MetricInterval > 0 { + ticker := time.NewTicker(s.MetricInterval) + defer ticker.Stop() + + gcn := gcnotifier.New() + defer gcn.Close() + + s.logger().Printf("runtime stats initializing (%s interval)", s.MetricInterval) + + for { + // Wait for tick or a close. + select { + case <-s.closing: + return + case <-gcn.AfterGC(): + // GC just ran + s.Index.Stats.Count("garbage_collection", 1) + s.logger().Printf("garbage collection complete") + case <-ticker.C: + } + + s.logger().Printf("runtime stats beginning") + + // TODO + s.Index.Stats.Gauge("goroutines", float64(runtime.NumGoroutine())) + + // Record successful sync in log. + s.logger().Printf("runtime stats complete") + } + } +} From 1ed4ecef25f81578b3759e248ab85d630ec3f65a Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 17:27:08 -0500 Subject: [PATCH 11/19] Don't try to create inverse views on Import() when inverseEnabled is false --- frame.go | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/frame.go b/frame.go index c70a07b68..414af55c6 100644 --- a/frame.go +++ b/frame.go @@ -500,18 +500,25 @@ func (f *Frame) Import(bitmapIDs, profileIDs []uint64, timestamps []*time.Time) dataByFragment[key] = data } - // Attach reversed bits to each inverse view. - for _, name := range inverse { - key := importKey{View: name, Slice: bitmapID / SliceWidth} - data := dataByFragment[key] - data.BitmapIDs = append(data.BitmapIDs, profileID) // reversed - data.ProfileIDs = append(data.ProfileIDs, bitmapID) // reversed - dataByFragment[key] = data + if f.inverseEnabled { + // Attach reversed bits to each inverse view. + for _, name := range inverse { + key := importKey{View: name, Slice: bitmapID / SliceWidth} + data := dataByFragment[key] + data.BitmapIDs = append(data.BitmapIDs, profileID) // reversed + data.ProfileIDs = append(data.ProfileIDs, bitmapID) // reversed + dataByFragment[key] = data + } } } // Import into each fragment. for key, data := range dataByFragment { + // Skip inverse data if inverse is not enabled. + if !f.inverseEnabled && IsInverseView(key.View) { + continue + } + // Re-sort data for inverse views. if IsInverseView(key.View) { sort.Sort(importBitSet{ From 5c50ab8f0e64e8f820d20a9edb66d5c748ab0c9c Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 19 Apr 2017 18:05:22 -0500 Subject: [PATCH 12/19] add a client test for imports to an inverseEnabled frame --- client_test.go | 55 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/client_test.go b/client_test.go index f808514a0..f78ee69c4 100644 --- a/client_test.go +++ b/client_test.go @@ -190,6 +190,61 @@ func TestClient_Import(t *testing.T) { } } +// Ensure client can bulk import data to an inverse frame. +func TestClient_ImportInverseEnabled(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + d := idx.MustCreateDBIfNotExists("d", pilosa.DBOptions{}) + frameOpts := pilosa.FrameOptions{ + InverseEnabled: true, + } + frame, err := d.CreateFrameIfNotExists("f", frameOpts) + if err != nil { + panic(err) + } + v, err := frame.CreateViewIfNotExists(pilosa.ViewInverse) + if err != nil { + panic(err) + } + f, err := v.CreateFragmentIfNotExists(0) + if err != nil { + panic(err) + } + + // Load bitmap into cache to ensure cache gets updated. + f.Bitmap(0) + + s := NewServer() + defer s.Close() + s.Handler.Host = s.Host() + s.Handler.Cluster = NewCluster(1) + s.Handler.Cluster.Nodes[0].Host = s.Host() + s.Handler.Index = idx.Index + + // Send import request. + c := MustNewClient(s.Host()) + if err := c.Import(context.Background(), "d", "f", 0, []pilosa.Bit{ + {BitmapID: 0, ProfileID: 1}, + {BitmapID: 0, ProfileID: 5}, + {BitmapID: 200, ProfileID: 5}, + {BitmapID: 200, ProfileID: 6}, + }); err != nil { + t.Fatal(err) + } + + // Verify data. + if a := f.Bitmap(1).Bits(); !reflect.DeepEqual(a, []uint64{0}) { + t.Fatalf("unexpected bits: %+v", a) + } + if a := f.Bitmap(5).Bits(); !reflect.DeepEqual(a, []uint64{0, 200}) { + t.Fatalf("unexpected bits: %+v", a) + } + if a := f.Bitmap(6).Bits(); !reflect.DeepEqual(a, []uint64{200}) { + t.Fatalf("unexpected bits: %+v", a) + } +} + // Ensure client backup and restore a frame. func TestClient_BackupRestore(t *testing.T) { idx := MustOpenIndex() From 749ffdcf65ad0898c1eca0258b10a498bc2da07a Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 10:41:17 -0500 Subject: [PATCH 13/19] adjusted DBOptions/FrameOptions logic and removed some unused tests --- db.go | 8 +++ frame.go | 23 +++++-- handler.go | 12 +--- handler_test.go | 54 --------------- index_test.go | 28 -------- internal/private.pb.go | 148 ++++++++++++++++++++--------------------- internal/private.proto | 4 +- 7 files changed, 105 insertions(+), 172 deletions(-) diff --git a/db.go b/db.go index d2ceaf13a..8bf33300e 100644 --- a/db.go +++ b/db.go @@ -536,6 +536,14 @@ type DBOptions struct { TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"` } +// Encode converts o into its internal representation. +func (o *DBOptions) Encode() *internal.DBMeta { + return &internal.DBMeta{ + ColumnLabel: o.ColumnLabel, + TimeQuantum: string(o.TimeQuantum), + } +} + // hasTime returns true if a contains a non-nil time. func hasTime(a []*time.Time) bool { for _, t := range a { diff --git a/frame.go b/frame.go index a497e6906..30fd41406 100644 --- a/frame.go +++ b/frame.go @@ -302,11 +302,11 @@ func (f *Frame) loadMeta() error { func (f *Frame) saveMeta() error { // Marshal metadata. buf, err := proto.Marshal(&internal.FrameMeta{ - TimeQuantum: string(f.timeQuantum), RowLabel: f.rowLabel, - CacheType: f.cacheType, InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), }) if err != nil { return err @@ -593,9 +593,11 @@ func encodeFrame(f *Frame) *internal.Frame { return &internal.Frame{ Name: f.name, Meta: &internal.FrameMeta{ - TimeQuantum: string(f.timeQuantum), - RowLabel: f.rowLabel, - CacheSize: f.cacheSize, + RowLabel: f.rowLabel, + InverseEnabled: f.inverseEnabled, + CacheType: f.cacheType, + CacheSize: f.cacheSize, + TimeQuantum: string(f.timeQuantum), }, } } @@ -627,6 +629,17 @@ type FrameOptions struct { TimeQuantum TimeQuantum `json:"timeQuantum,omitempty"` } +// Encode converts o into its internal representation. +func (o *FrameOptions) Encode() *internal.FrameMeta { + return &internal.FrameMeta{ + RowLabel: o.RowLabel, + InverseEnabled: o.InverseEnabled, + CacheType: o.CacheType, + CacheSize: o.CacheSize, + TimeQuantum: string(o.TimeQuantum), + } +} + // importBitSet represents slices of row and column ids. // This is used to sort data during import. type importBitSet struct { diff --git a/handler.go b/handler.go index d9c79eeba..8b79267bf 100644 --- a/handler.go +++ b/handler.go @@ -354,11 +354,8 @@ func (h *Handler) handlePostDB(w http.ResponseWriter, r *http.Request) { // Send the create database message to all nodes. err = h.Broadcaster.SendSync( &internal.CreateDBMessage{ - DB: dbName, - Meta: &internal.DBMeta{ - ColumnLabel: req.Options.ColumnLabel, - TimeQuantum: string(req.Options.TimeQuantum), - }, + DB: dbName, + Meta: req.Options.Encode(), }) if err != nil { h.logger().Printf("problem sending CreateDB message: %s", err) @@ -508,10 +505,7 @@ func (h *Handler) handlePostFrame(w http.ResponseWriter, r *http.Request) { &internal.CreateFrameMessage{ DB: dbName, Frame: frameName, - Meta: &internal.FrameMeta{ - RowLabel: req.Options.RowLabel, - TimeQuantum: string(req.Options.TimeQuantum), - }, + Meta: req.Options.Encode(), }) if err != nil { h.logger().Printf("problem sending CreateFrame message: %s", err) diff --git a/handler_test.go b/handler_test.go index 478720e23..049fdc036 100644 --- a/handler_test.go +++ b/handler_test.go @@ -874,57 +874,3 @@ func MustReadAll(r io.Reader) []byte { } return buf } - -/* -// TODO: move this test to messenger.go (with NewServer()) - -// Ensure that an HTTP message sent to the cluster reaches all nodes. -func TestHTTPNodeSet_Base(t *testing.T) { - - // servers - s1 := NewServer() - s1.Messenger = pilosa.NewMessenger() - n1 := NewHTTPMessageBroker() - n1.messenger = s1.Messenger - s1.Messenger.Broker = n1 - - s2 := NewServer() - s2.Messenger = pilosa.NewMessenger() - n2 := NewHTTPMessageBroker() - n2.messenger = s2.Messenger - s2.Messenger.Broker = n2 - - s3 := NewServer() - s3.Messenger = pilosa.NewMessenger() - n3 := NewHTTPMessageBroker() - n3.messenger = s3.Messenger - s3.Messenger.Broker = n3 - - nodes := []*pilosa.Node{ - {Host: s1.Host()}, - {Host: s2.Host()}, - {Host: s3.Host()}, - } - - // message - msg := &internal.CreateSliceMessage{ - DB: "d", - Slice: 8, - } - - // send message - if err := s1.Messenger.SendMessage(msg, ""); err != nil { - t.Fatalf("failure sending message: %s", err) - } - - if !reflect.DeepEqual(mb1.messageReceived, msg) { - t.Fatalf("unexpected message received by node1: %s", mb1.messageReceived) - } - if !reflect.DeepEqual(mb2.messageReceived, msg) { - t.Fatalf("unexpected message received by node2: %s", mb2.messageReceived) - } - if !reflect.DeepEqual(mb3.messageReceived, msg) { - t.Fatalf("unexpected message received by node3: %s", mb3.messageReceived) - } -} -*/ diff --git a/index_test.go b/index_test.go index c09fe69f5..4a1be4cf2 100644 --- a/index_test.go +++ b/index_test.go @@ -162,34 +162,6 @@ func TestIndexSyncer_SyncIndex(t *testing.T) { } } -/* TODO: move this to messenger.go -// Ensure index can handle Messenger messages. -func TestIndex_HandleMessage(t *testing.T) { - // Create a local index. - idx0 := MustOpenIndex() - defer idx0.Close() - - idx0.MustCreateDBIfNotExists("d", pilosa.DBOptions{}) - - msg0 := &internal.CreateSliceMessage{ - DB: "d", - Slice: 8, - } - idx0.HandleMessage(msg0) - if ms := idx0.MaxSlices(); !reflect.DeepEqual(ms, map[string]uint64{"d": 8}) { - t.Fatalf("unexpected max slice: %s", ms) - } - - msg1 := &internal.DeleteDBMessage{ - DB: "d", - } - idx0.HandleMessage(msg1) - if ms := idx0.MaxSlices(); !reflect.DeepEqual(ms, map[string]uint64{}) { - t.Fatalf("unexpected delete db: %s", ms) - } -} -*/ - // Index is a test wrapper for pilosa.Index. type Index struct { *pilosa.Index diff --git a/internal/private.pb.go b/internal/private.pb.go index 5ceb800c2..0423cfa08 100644 --- a/internal/private.pb.go +++ b/internal/private.pb.go @@ -45,8 +45,8 @@ var _ = math.Inf const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package type DBMeta struct { - TimeQuantum string `protobuf:"bytes,1,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` - ColumnLabel string `protobuf:"bytes,2,opt,name=ColumnLabel,proto3" json:"ColumnLabel,omitempty"` + ColumnLabel string `protobuf:"bytes,1,opt,name=ColumnLabel,proto3" json:"ColumnLabel,omitempty"` + TimeQuantum string `protobuf:"bytes,2,opt,name=TimeQuantum,proto3" json:"TimeQuantum,omitempty"` } func (m *DBMeta) Reset() { *m = DBMeta{} } @@ -281,18 +281,18 @@ func (m *DBMeta) MarshalTo(dAtA []byte) (int, error) { _ = i var l int _ = l - if len(m.TimeQuantum) > 0 { - dAtA[i] = 0xa - i++ - i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) - i += copy(dAtA[i:], m.TimeQuantum) - } if len(m.ColumnLabel) > 0 { - dAtA[i] = 0x12 + dAtA[i] = 0xa i++ i = encodeVarintPrivate(dAtA, i, uint64(len(m.ColumnLabel))) i += copy(dAtA[i:], m.ColumnLabel) } + if len(m.TimeQuantum) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintPrivate(dAtA, i, uint64(len(m.TimeQuantum))) + i += copy(dAtA[i:], m.TimeQuantum) + } return i, nil } @@ -852,11 +852,11 @@ func encodeVarintPrivate(dAtA []byte, offset int, v uint64) int { func (m *DBMeta) Size() (n int) { var l int _ = l - l = len(m.TimeQuantum) + l = len(m.ColumnLabel) if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } - l = len(m.ColumnLabel) + l = len(m.TimeQuantum) if l > 0 { n += 1 + l + sovPrivate(uint64(l)) } @@ -1137,35 +1137,6 @@ func (m *DBMeta) Unmarshal(dAtA []byte) error { } switch fieldNum { case 1: - if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field TimeQuantum", wireType) - } - var stringLen uint64 - for shift := uint(0); ; shift += 7 { - if shift >= 64 { - return ErrIntOverflowPrivate - } - if iNdEx >= l { - return io.ErrUnexpectedEOF - } - b := dAtA[iNdEx] - iNdEx++ - stringLen |= (uint64(b) & 0x7F) << shift - if b < 0x80 { - break - } - } - intStringLen := int(stringLen) - if intStringLen < 0 { - return ErrInvalidLengthPrivate - } - postIndex := iNdEx + intStringLen - if postIndex > l { - return io.ErrUnexpectedEOF - } - m.TimeQuantum = string(dAtA[iNdEx:postIndex]) - iNdEx = postIndex - case 2: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field ColumnLabel", wireType) } @@ -1194,6 +1165,35 @@ func (m *DBMeta) Unmarshal(dAtA []byte) error { } m.ColumnLabel = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field TimeQuantum", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPrivate + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPrivate + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.TimeQuantum = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPrivate(dAtA[iNdEx:]) @@ -3146,43 +3146,43 @@ var ( func init() { proto.RegisterFile("private.proto", fileDescriptorPrivate) } var fileDescriptorPrivate = []byte{ - // 603 bytes of a gzipped FileDescriptorProto + // 600 bytes of a gzipped FileDescriptorProto 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x94, 0x54, 0xcd, 0x4e, 0x14, 0x41, 0x10, 0x76, 0x7e, 0x20, 0x4c, 0x21, 0xcb, 0xd2, 0x7a, 0x98, 0x10, 0x32, 0x59, 0x3b, 0x2a, 0xc4, 0x03, 0x07, 0xbc, 0x18, 0xe2, 0x69, 0x18, 0x14, 0x12, 0x20, 0xd2, 0x8b, 0xde, 0x7b, 0x97, 0x52, 0x27, 0x3b, 0x7f, 0xce, 0xf4, 0x2e, 0xac, 0x57, 0x5f, 0xc2, 0xc4, 0x67, 0xf0, 0x3d, 0x3c, 0xfa, - 0x08, 0x66, 0x7d, 0x11, 0xd3, 0xdd, 0xf3, 0xe7, 0xb2, 0xf8, 0x73, 0xeb, 0xfa, 0xaa, 0xfa, 0xab, - 0xaf, 0xbe, 0xa9, 0x1e, 0x58, 0xcb, 0xf2, 0x70, 0xc2, 0x05, 0xee, 0x66, 0x79, 0x2a, 0x52, 0xb2, + 0x08, 0x66, 0x7d, 0x11, 0xd3, 0xdd, 0xf3, 0xe7, 0xb2, 0xf8, 0x73, 0xeb, 0xfa, 0xaa, 0xea, 0xab, + 0xaf, 0xbf, 0xa9, 0x1e, 0x58, 0xcb, 0xf2, 0x70, 0xc2, 0x05, 0xee, 0x66, 0x79, 0x2a, 0x52, 0xb2, 0x12, 0x26, 0x02, 0xf3, 0x84, 0x47, 0xf4, 0x04, 0x96, 0x03, 0xff, 0x14, 0x05, 0x27, 0x3d, 0x58, - 0xbd, 0x08, 0x63, 0x3c, 0x1f, 0xf3, 0x44, 0x8c, 0x63, 0xd7, 0xe8, 0x19, 0x3b, 0x0e, 0x6b, 0x43, - 0xb2, 0xe2, 0x20, 0x8d, 0xc6, 0x71, 0x72, 0xc2, 0x07, 0x18, 0xb9, 0xa6, 0xae, 0x68, 0x41, 0xf4, - 0xab, 0x01, 0xce, 0x8b, 0x9c, 0xc7, 0xa8, 0x18, 0x37, 0x61, 0x85, 0xa5, 0x57, 0xba, 0x58, 0xd3, - 0xd5, 0x31, 0x79, 0x0c, 0x9d, 0xe3, 0x64, 0x82, 0x79, 0x81, 0x87, 0x09, 0x1f, 0x44, 0x78, 0xa9, - 0xe8, 0x56, 0xd8, 0x1c, 0x4a, 0xb6, 0xc0, 0x39, 0xe0, 0xc3, 0xf7, 0x78, 0x31, 0xcd, 0xd0, 0xb5, - 0x14, 0x49, 0x03, 0xd4, 0xd9, 0x7e, 0xf8, 0x11, 0x5d, 0xbb, 0x67, 0xec, 0xac, 0xb1, 0x06, 0x98, - 0x9f, 0x68, 0xe9, 0xc6, 0x44, 0x94, 0x42, 0xe7, 0x38, 0xce, 0xd2, 0x5c, 0x30, 0x2c, 0xb2, 0x34, - 0x29, 0x90, 0x74, 0xc1, 0x3a, 0xcc, 0xf3, 0x52, 0xae, 0x3c, 0xd2, 0x6b, 0xe8, 0xfa, 0x51, 0x3a, - 0x1c, 0x05, 0x5c, 0x70, 0x86, 0x1f, 0xc6, 0x58, 0x08, 0xd2, 0x01, 0x33, 0xf0, 0xcb, 0x22, 0x33, - 0xf0, 0xc9, 0x7d, 0x58, 0x52, 0x63, 0x97, 0x9e, 0xe8, 0x40, 0xa2, 0xea, 0xa6, 0xd2, 0x6d, 0x33, - 0x1d, 0x48, 0xb4, 0x1f, 0x85, 0x43, 0xad, 0xd7, 0x66, 0x3a, 0x20, 0x04, 0xec, 0x37, 0x21, 0x5e, - 0x95, 0x22, 0xd5, 0x99, 0x9e, 0xc3, 0x46, 0xab, 0x73, 0x29, 0x70, 0x0b, 0x1c, 0x3f, 0x14, 0x31, - 0xcf, 0x8e, 0x83, 0xc2, 0x35, 0x7a, 0xd6, 0x8e, 0xcd, 0x1a, 0x80, 0x78, 0x00, 0xaf, 0xf2, 0xf4, - 0x6d, 0x18, 0xa1, 0x4c, 0x9b, 0x2a, 0xdd, 0x42, 0xe8, 0x23, 0x58, 0x52, 0xfe, 0xfc, 0x99, 0x86, - 0x7e, 0x31, 0x60, 0xe3, 0x94, 0x5f, 0x2b, 0x69, 0x45, 0xdd, 0xfa, 0x08, 0x9c, 0x1a, 0x54, 0x77, - 0x56, 0xf7, 0x9e, 0xec, 0x56, 0x9b, 0xb4, 0x7b, 0xa3, 0xbe, 0x41, 0x0e, 0x13, 0x91, 0x4f, 0x59, - 0x73, 0x79, 0xf3, 0x39, 0x74, 0x7e, 0x4f, 0x4a, 0xdf, 0x47, 0x38, 0xad, 0x7c, 0x1f, 0xe1, 0x54, - 0xfa, 0x34, 0xe1, 0xd1, 0x58, 0x7b, 0x6a, 0x33, 0x1d, 0xec, 0x9b, 0xcf, 0x0c, 0xba, 0x0f, 0xe4, - 0x20, 0x47, 0x2e, 0x50, 0x11, 0x9c, 0x62, 0x51, 0xf0, 0x77, 0xb8, 0xe8, 0x9b, 0x68, 0x9f, 0xcd, - 0x96, 0xcf, 0xf4, 0x01, 0xac, 0x07, 0x18, 0xa1, 0x40, 0xb9, 0xf5, 0x0b, 0x2f, 0xd2, 0x97, 0xb0, - 0xae, 0xe9, 0x6f, 0x2d, 0x21, 0x0f, 0xc1, 0x96, 0x1b, 0xae, 0xa8, 0x57, 0xf7, 0xba, 0x8d, 0x09, - 0xfa, 0x2d, 0x31, 0x95, 0xa5, 0xc3, 0x4a, 0x67, 0xf9, 0x24, 0x6e, 0xd5, 0xb9, 0x60, 0x77, 0xb6, - 0xcb, 0x0e, 0x96, 0xea, 0x70, 0xaf, 0xe9, 0x50, 0x3f, 0xaf, 0xb2, 0xc9, 0x3e, 0x10, 0x3d, 0xd0, - 0xff, 0x37, 0xa1, 0x41, 0x89, 0xca, 0xed, 0x3b, 0x93, 0x59, 0x7d, 0x41, 0x9d, 0x6b, 0x05, 0xe6, - 0xdf, 0x14, 0x7c, 0x32, 0x64, 0xb3, 0x85, 0x1c, 0xff, 0xe4, 0x93, 0xfc, 0x4f, 0x54, 0xdb, 0x50, - 0x3e, 0x95, 0x3a, 0x26, 0xdb, 0xb0, 0xac, 0xfa, 0x15, 0xae, 0xad, 0x16, 0x6e, 0x7d, 0x4e, 0x07, - 0x2b, 0xd3, 0xf4, 0x35, 0x38, 0x67, 0xe9, 0x25, 0xf6, 0x05, 0x17, 0x6a, 0x9e, 0xa3, 0xb4, 0x10, - 0x95, 0x16, 0x79, 0x56, 0xfb, 0x20, 0x93, 0x95, 0x05, 0xba, 0xd2, 0x03, 0x2b, 0xf0, 0x0b, 0xd7, - 0x52, 0xe4, 0x77, 0xdb, 0x02, 0x99, 0x4c, 0xf8, 0xdd, 0x6f, 0x33, 0xcf, 0xf8, 0x3e, 0xf3, 0x8c, - 0x1f, 0x33, 0xcf, 0xf8, 0xfc, 0xd3, 0xbb, 0x33, 0x58, 0x56, 0xbf, 0xd0, 0xa7, 0xbf, 0x02, 0x00, - 0x00, 0xff, 0xff, 0xc3, 0x9a, 0x29, 0xff, 0x53, 0x05, 0x00, 0x00, + 0x3d, 0x48, 0xa3, 0x71, 0x9c, 0x9c, 0xf0, 0x01, 0x46, 0xae, 0xd1, 0x33, 0x76, 0x1c, 0xd6, 0x86, + 0x64, 0xc5, 0x45, 0x18, 0xe3, 0xf9, 0x98, 0x27, 0x62, 0x1c, 0xbb, 0xa6, 0xae, 0x68, 0x41, 0xf4, + 0xab, 0x01, 0xce, 0x8b, 0x9c, 0xc7, 0xa8, 0x18, 0x37, 0x61, 0x85, 0xa5, 0x57, 0x6d, 0xba, 0x3a, + 0x26, 0x8f, 0xa1, 0x73, 0x9c, 0x4c, 0x30, 0x2f, 0xf0, 0x30, 0xe1, 0x83, 0x08, 0x2f, 0x15, 0xdd, + 0x0a, 0x9b, 0x43, 0xc9, 0x16, 0x38, 0x07, 0x7c, 0xf8, 0x1e, 0x2f, 0xa6, 0x19, 0xba, 0x96, 0x22, + 0x69, 0x80, 0x3a, 0xdb, 0x0f, 0x3f, 0xa2, 0x6b, 0xf7, 0x8c, 0x9d, 0x35, 0xd6, 0x00, 0xf3, 0x7a, + 0x97, 0x6e, 0xea, 0xa5, 0xd0, 0x39, 0x8e, 0xb3, 0x34, 0x17, 0x0c, 0x8b, 0x2c, 0x4d, 0x0a, 0x24, + 0x5d, 0xb0, 0x0e, 0xf3, 0xbc, 0x94, 0x2b, 0x8f, 0xf4, 0x1a, 0xba, 0x7e, 0x94, 0x0e, 0x47, 0x01, + 0x17, 0x9c, 0xe1, 0x87, 0x31, 0x16, 0x82, 0x74, 0xc0, 0x0c, 0xfc, 0xb2, 0xc8, 0x0c, 0x7c, 0x72, + 0x1f, 0x96, 0xd4, 0xb5, 0x4b, 0x4f, 0x74, 0x20, 0x51, 0xd5, 0xa9, 0x74, 0xdb, 0x4c, 0x07, 0x12, + 0xed, 0x47, 0xe1, 0x50, 0xeb, 0xb5, 0x99, 0x0e, 0x08, 0x01, 0xfb, 0x4d, 0x88, 0x57, 0xa5, 0x48, + 0x75, 0xa6, 0xe7, 0xb0, 0xd1, 0x9a, 0x5c, 0x0a, 0xdc, 0x02, 0xc7, 0x0f, 0x45, 0xcc, 0xb3, 0xe3, + 0xa0, 0x70, 0x8d, 0x9e, 0xb5, 0x63, 0xb3, 0x06, 0x20, 0x1e, 0xc0, 0xab, 0x3c, 0x7d, 0x1b, 0x46, + 0x28, 0xd3, 0xa6, 0x4a, 0xb7, 0x10, 0xfa, 0x08, 0x96, 0x94, 0x3f, 0x7f, 0xa6, 0xa1, 0x5f, 0x0c, + 0xd8, 0x38, 0xe5, 0xd7, 0x4a, 0x5a, 0x51, 0x8f, 0x3e, 0x02, 0xa7, 0x06, 0x55, 0xcf, 0xea, 0xde, + 0x93, 0xdd, 0x6a, 0x93, 0x76, 0x6f, 0xd4, 0x37, 0xc8, 0x61, 0x22, 0xf2, 0x29, 0x6b, 0x9a, 0x37, + 0x9f, 0x43, 0xe7, 0xf7, 0xa4, 0xf4, 0x7d, 0x84, 0xd3, 0xca, 0xf7, 0x11, 0x4e, 0xa5, 0x4f, 0x13, + 0x1e, 0x8d, 0xb5, 0xa7, 0x36, 0xd3, 0xc1, 0xbe, 0xf9, 0xcc, 0xa0, 0xfb, 0x40, 0x0e, 0x72, 0xe4, + 0x02, 0x15, 0xc1, 0x29, 0x16, 0x05, 0x7f, 0x87, 0x8b, 0xbe, 0x89, 0xf6, 0xd9, 0x6c, 0xf9, 0x4c, + 0x1f, 0xc0, 0x7a, 0x80, 0x11, 0x0a, 0x94, 0x5b, 0xbf, 0xb0, 0x91, 0xbe, 0x84, 0x75, 0x4d, 0x7f, + 0x6b, 0x09, 0x79, 0x08, 0xb6, 0xdc, 0x70, 0x45, 0xbd, 0xba, 0xd7, 0x6d, 0x4c, 0xd0, 0x6f, 0x89, + 0xa9, 0x2c, 0x1d, 0x56, 0x3a, 0xcb, 0x27, 0x71, 0xab, 0xce, 0x05, 0xbb, 0xb3, 0x5d, 0x4e, 0xb0, + 0xd4, 0x84, 0x7b, 0xcd, 0x84, 0xfa, 0x79, 0x95, 0x43, 0xf6, 0x81, 0xe8, 0x0b, 0xfd, 0xff, 0x10, + 0x1a, 0x94, 0xa8, 0xdc, 0xbe, 0x33, 0x99, 0xd5, 0x0d, 0xea, 0x5c, 0x2b, 0x30, 0xff, 0xa6, 0xe0, + 0x93, 0x21, 0x87, 0x2d, 0xe4, 0xf8, 0x27, 0x9f, 0xe4, 0x7f, 0xa2, 0xda, 0x86, 0xf2, 0xa9, 0xd4, + 0x31, 0xd9, 0x86, 0x65, 0x35, 0xaf, 0x70, 0x6d, 0xb5, 0x70, 0xeb, 0x73, 0x3a, 0x58, 0x99, 0xa6, + 0xaf, 0xc1, 0x39, 0x4b, 0x2f, 0xb1, 0x2f, 0xb8, 0x50, 0xf7, 0x39, 0x4a, 0x0b, 0x51, 0x69, 0x91, + 0x67, 0xb5, 0x0f, 0x32, 0x59, 0x59, 0xa0, 0x2b, 0x3d, 0xb0, 0x02, 0xbf, 0x70, 0x2d, 0x45, 0x7e, + 0xb7, 0x2d, 0x90, 0xc9, 0x84, 0xdf, 0xfd, 0x36, 0xf3, 0x8c, 0xef, 0x33, 0xcf, 0xf8, 0x31, 0xf3, + 0x8c, 0xcf, 0x3f, 0xbd, 0x3b, 0x83, 0x65, 0xf5, 0x0b, 0x7d, 0xfa, 0x2b, 0x00, 0x00, 0xff, 0xff, + 0x3a, 0x23, 0x0f, 0xb4, 0x53, 0x05, 0x00, 0x00, } diff --git a/internal/private.proto b/internal/private.proto index 40e83eb6b..ec060e355 100644 --- a/internal/private.proto +++ b/internal/private.proto @@ -3,8 +3,8 @@ syntax = "proto3"; package internal; message DBMeta { - string TimeQuantum = 1; - string ColumnLabel = 2; + string ColumnLabel = 1; + string TimeQuantum = 2; } message FrameMeta { From 070a27aa317a784f778e0509cd051c56e21016e8 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Thu, 20 Apr 2017 14:23:24 -0500 Subject: [PATCH 14/19] Store the node state information in Cluster Access the overall cluster state info via the /status endpoint --- cluster.go | 39 +++++++++++++++++++++++++++++++++++++++ handler.go | 28 ++++++++++++++++++++++------ server.go | 31 ++++++++++++++++++++++--------- 3 files changed, 83 insertions(+), 15 deletions(-) diff --git a/cluster.go b/cluster.go index f92b8c6e9..0e874e06e 100644 --- a/cluster.go +++ b/cluster.go @@ -3,6 +3,9 @@ package pilosa import ( "encoding/binary" "hash/fnv" + "sync" + + "github.com/pilosa/pilosa/internal" ) const ( @@ -86,6 +89,7 @@ func (a Nodes) Clone() []*Node { // Cluster represents a collection of nodes. type Cluster struct { + mu sync.Mutex Nodes []*Node NodeSet NodeSet @@ -97,6 +101,9 @@ type Cluster struct { // The number of replicas a partition has. ReplicaN int + + // Current state of nodes in the cluster + NodeState map[string]*internal.NodeState } // NewCluster returns a new instance of Cluster with defaults. @@ -105,6 +112,7 @@ func NewCluster() *Cluster { Hasher: &jmphasher{}, PartitionN: DefaultPartitionN, ReplicaN: DefaultReplicaN, + NodeState: make(map[string]*internal.NodeState), } } @@ -190,6 +198,37 @@ func (c *Cluster) PartitionNodes(partitionID int) []*Node { return nodes } +// OwnsSlices find the set of slices owned by the node per DB +func (c *Cluster) OwnsSlices(db string, maxSlice uint64, host string) []uint64 { + var slices []uint64 + for i := uint64(0); i <= maxSlice; i++ { + p := c.Partition(db, i) + // Determine primary owner node. + index := c.Hasher.Hash(uint64(p), len(c.Nodes)) + if c.Nodes[index].Host == host { + slices = append(slices, i) + } + + } + return slices +} + +// SetNodeState stores the remote node states transmitted through gossip +func (c *Cluster) SetNodeState(state *internal.NodeState) { + c.mu.Lock() + defer c.mu.Unlock() + + c.NodeState[state.Host] = state +} + +// GetNodeState stores the remote node states transmitted through gossip +func (c *Cluster) GetNodeState(host string) *internal.NodeState { + c.mu.Lock() + defer c.mu.Unlock() + + return c.NodeState[host] +} + // Hasher represents an interface to hash integers into buckets. type Hasher interface { // Hashes the key into a number between [0,N). diff --git a/handler.go b/handler.go index a6309e44a..a72aa701d 100644 --- a/handler.go +++ b/handler.go @@ -23,10 +23,18 @@ import ( "github.com/pilosa/pilosa/pql" ) +// ServerHandler a method to update the local node's state information +// this is used to handle the cluster status request and append the +// local node's state with the cluster state gathered via Gossip +type ServerHandler interface { + HandleStateRequest() error +} + // Handler represents an HTTP handler. type Handler struct { - Index *Index - Broadcaster Broadcaster + Index *Index + Broadcaster Broadcaster + ServerHandler ServerHandler // Local hostname & cluster configuration. Host string @@ -83,6 +91,7 @@ func NewRouter(handler *Handler) *mux.Router { router.HandleFunc("/nodes", handler.handleGetNodes).Methods("GET") router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET") router.HandleFunc("/slices/max", handler.handleGetSliceMax).Methods("GET") + router.HandleFunc("/status", handler.handleGetStatus).Methods("GET") router.HandleFunc("/version", handler.handleGetVersion).Methods("GET") // TODO: Apply MethodNotAllowed statuses to all endpoints. @@ -112,12 +121,17 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { } } -// handleGetStatus handles GET /status requests. func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) { + // Compute my local state + fmt.Println("Call interface") + h.ServerHandler.HandleStateRequest() + if err := json.NewEncoder(w).Encode(getStatusResponse{ - Health: h.Cluster.Health(), + Health: h.Cluster.NodeState, + Version: h.Version, + Replicas: h.Cluster.ReplicaN, }); err != nil { - h.logger().Printf("write status response error: %s", err) + h.logger().Printf("Node State Error: %s", err) } } @@ -126,7 +140,9 @@ type getSchemaResponse struct { } type getStatusResponse struct { - Health map[string]string `json:"health"` + Health map[string]*internal.NodeState `json:"health"` + Version string `json:"version"` + Replicas int } // handlePostQuery handles /query requests. diff --git a/server.go b/server.go index 68337049d..f7e0b232c 100644 --- a/server.go +++ b/server.go @@ -71,6 +71,7 @@ func NewServer() *Server { } s.Handler.Index = s.Index + s.Handler.ServerHandler = s return s } @@ -241,6 +242,11 @@ func (s *Server) monitorMaxSlices() { } } +func (s *Server) HandleStateRequest() error { + _, err := s.LocalState() + return err +} + // LocalState returns the state of the local node as well as the // index (dbs/frames) according to the local node. // In a gossip implementation, memberlist.Delegate.LocalState() uses this. @@ -248,11 +254,22 @@ func (s *Server) LocalState() (proto.Message, error) { if s.Index == nil { return nil, errors.New("Server.Index is nil.") } - return &internal.NodeState{ + + // Get Node DB Slices + for _, db := range s.Index.DBs() { + maxSlice := db.MaxSlice() + slices := s.Cluster.OwnsSlices(db.name, maxSlice, s.Host) + fmt.Println("Slices ", slices) + } + + ns := internal.NodeState{ Host: s.Host, State: "OK", // TODO: make this work, pull from s.Cluster.Node DBs: encodeDBs(s.Index.DBs()), - }, nil + } + + s.Cluster.SetNodeState(&ns) + return &ns, nil } func (s *Server) ReceiveMessage(pb proto.Message) error { @@ -296,7 +313,8 @@ func (s *Server) HandleRemoteState(pb proto.Message) error { } func (s *Server) mergeRemoteState(ns *internal.NodeState) error { - // TODO: update some node state value in the cluster (it should be in cluster.node i guess) + // store this node's state in the cluster node map + s.Cluster.SetNodeState(ns) // Create databases that don't exist. for _, db := range ns.DBs { @@ -392,13 +410,8 @@ func (s *Server) monitorRuntime() { case <-ticker.C: } - s.logger().Printf("runtime stats beginning") - - // TODO + // Record the number of go routines s.Index.Stats.Gauge("goroutines", float64(runtime.NumGoroutine())) - - // Record successful sync in log. - s.logger().Printf("runtime stats complete") } } } From 281e6e8e0fd40771d48459f64c6e783c92276b99 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 15:14:11 -0500 Subject: [PATCH 15/19] remove leftover Receive() method from gossip implementation --- gossip/gossip.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/gossip/gossip.go b/gossip/gossip.go index 050435441..772892926 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -154,13 +154,6 @@ func (g *GossipNodeSet) SendAsync(pb proto.Message) error { return nil } -func (g *GossipNodeSet) Receive(pb proto.Message) error { - if err := g.handler.ReceiveMessage(pb); err != nil { - return err - } - return nil -} - // implementation of the memberlist.Delegate interface func (g *GossipNodeSet) NodeMeta(limit int) []byte { return []byte{} @@ -172,7 +165,7 @@ func (g *GossipNodeSet) NotifyMsg(b []byte) { g.logger().Printf("unmarshal message error: %s", err) return } - if err := g.Receive(m); err != nil { + if err := g.handler.ReceiveMessage(m); err != nil { g.logger().Printf("receive message error: %s", err) return } From 3f3c40e2412d2e4fbd22a70ebd791e0f8fa9c258 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 20 Apr 2017 15:45:00 -0500 Subject: [PATCH 16/19] group Server interface implementation function together --- server.go | 30 ++++++++++++++++-------------- 1 file changed, 16 insertions(+), 14 deletions(-) diff --git a/server.go b/server.go index e93b8e111..567461ccc 100644 --- a/server.go +++ b/server.go @@ -236,20 +236,7 @@ func (s *Server) monitorMaxSlices() { } } -// LocalState returns the state of the local node as well as the -// index (dbs/frames) according to the local node. -// In a gossip implementation, memberlist.Delegate.LocalState() uses this. -func (s *Server) LocalState() (proto.Message, error) { - if s.Index == nil { - return nil, errors.New("Server.Index is nil.") - } - return &internal.NodeState{ - Host: s.Host, - State: "OK", // TODO: make this work, pull from s.Cluster.Node - DBs: encodeDBs(s.Index.DBs()), - }, nil -} - +// ReceiveMessage represents an implementation of BroadcastHandler. func (s *Server) ReceiveMessage(pb proto.Message) error { switch obj := pb.(type) { case *internal.CreateSliceMessage: @@ -284,6 +271,21 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return nil } +// Server implements gossip.StateHandler. +// LocalState returns the state of the local node as well as the +// index (dbs/frames) according to the local node. +// In a gossip implementation, memberlist.Delegate.LocalState() uses this. +func (s *Server) LocalState() (proto.Message, error) { + if s.Index == nil { + return nil, errors.New("Server.Index is nil.") + } + return &internal.NodeState{ + Host: s.Host, + State: "OK", // TODO: make this work, pull from s.Cluster.Node + DBs: encodeDBs(s.Index.DBs()), + }, nil +} + // HandleRemoteState receives incoming NodeState from remote nodes. func (s *Server) HandleRemoteState(pb proto.Message) error { return s.mergeRemoteState(pb.(*internal.NodeState)) From 81525f2d4ab7d9a5ddd3e45ab771c58363e95548 Mon Sep 17 00:00:00 2001 From: Michael Baird Date: Thu, 20 Apr 2017 16:52:05 -0500 Subject: [PATCH 17/19] move the StateHandler interface from the Gossip package back to Pilosa and us this interface for the Handler to access the LocalState --- gossip/gossip.go | 12 ++---------- handler.go | 11 ++--------- server.go | 18 ++++++++++-------- 3 files changed, 14 insertions(+), 27 deletions(-) diff --git a/gossip/gossip.go b/gossip/gossip.go index 772892926..513bd03d1 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -14,14 +14,6 @@ import ( "github.com/pilosa/pilosa/internal" ) -// StateHandler specifies two methods which an object must implement to share -// state in the cluster. These are used by the GossipNodeSet to implement the -// LocalState and MergeRemoteState methods of memberlist.Delegate -type StateHandler interface { - LocalState() (proto.Message, error) - HandleRemoteState(proto.Message) error -} - // GossipNodeSet represents a gossip implementation of NodeSet using memberlist // GossipNodeSet also represents a gossip implementation of pilosa.Broadcaster // GossipNodeSet also represents an implementation of memberlist.Delegate @@ -31,7 +23,7 @@ type GossipNodeSet struct { broadcasts *memberlist.TransmitLimitedQueue - stateHandler StateHandler + stateHandler pilosa.StateHandler config *GossipConfig // The writer for any logging. @@ -89,7 +81,7 @@ type GossipConfig struct { } // NewGossipNodeSet returns a new instance of GossipNodeSet. -func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, sh StateHandler) *GossipNodeSet { +func NewGossipNodeSet(name string, gossipHost string, gossipPort int, gossipSeed string, sh pilosa.StateHandler) *GossipNodeSet { g := &GossipNodeSet{ LogOutput: os.Stderr, } diff --git a/handler.go b/handler.go index 9b110c609..a8bea8de1 100644 --- a/handler.go +++ b/handler.go @@ -23,18 +23,11 @@ import ( "github.com/pilosa/pilosa/pql" ) -// ServerHandler a method to update the local node's state information -// this is used to handle the cluster status request and append the -// local node's state with the cluster state gathered via Gossip -type ServerHandler interface { - HandleStateRequest() error -} - // Handler represents an HTTP handler. type Handler struct { Index *Index Broadcaster Broadcaster - ServerHandler ServerHandler + ServerHandler StateHandler // Local hostname & cluster configuration. Host string @@ -124,7 +117,7 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) { // Compute my local state fmt.Println("Call interface") - h.ServerHandler.HandleStateRequest() + h.ServerHandler.LocalState() if err := json.NewEncoder(w).Encode(getStatusResponse{ Health: h.Cluster.NodeState, diff --git a/server.go b/server.go index f7078e4a0..b22744bae 100644 --- a/server.go +++ b/server.go @@ -26,6 +26,14 @@ const ( DefaultPollingInterval = 60 * time.Second ) +// StateHandler specifies two methods which an object must implement to share +// state in the cluster. These are used by the GossipNodeSet to implement the +// LocalState and MergeRemoteState methods of memberlist.Delegate +type StateHandler interface { + LocalState() (proto.Message, error) + HandleRemoteState(proto.Message) error +} + // Server represents an index wrapped by a running HTTP server. type Server struct { ln net.Listener @@ -242,11 +250,6 @@ func (s *Server) monitorMaxSlices() { } } -func (s *Server) HandleStateRequest() error { - _, err := s.LocalState() - return err -} - func (s *Server) ReceiveMessage(pb proto.Message) error { switch obj := pb.(type) { case *internal.CreateSliceMessage: @@ -281,13 +284,13 @@ func (s *Server) ReceiveMessage(pb proto.Message) error { return nil } -// Server implements gossip.StateHandler. // LocalState returns the state of the local node as well as the // index (dbs/frames) according to the local node. +// Server implements gossip.StateHandler. // In a gossip implementation, memberlist.Delegate.LocalState() uses this. func (s *Server) LocalState() (proto.Message, error) { if s.Index == nil { - return nil, errors.New("Server.Index is nil.") + return nil, errors.New("Server.Index is nil") } // Get Node DB Slices @@ -406,7 +409,6 @@ func (s *Server) monitorRuntime() { case <-gcn.AfterGC(): // GC just ran s.Index.Stats.Count("garbage_collection", 1) - s.logger().Printf("garbage collection complete") case <-ticker.C: } From 4f07113f65f268cd355c22cb217ffcae32f04f2e Mon Sep 17 00:00:00 2001 From: Linh Vo Date: Thu, 20 Apr 2017 21:37:31 -0500 Subject: [PATCH 18/19] readd stastD report with new branch --- datadog/datadog.go | 8 +++ db.go | 30 ++++----- executor.go | 10 ++- fragment.go | 2 +- frame.go | 32 ++++----- index.go | 2 +- stats.go | 16 +++++ stats_test.go | 163 +++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 228 insertions(+), 35 deletions(-) create mode 100644 stats_test.go diff --git a/datadog/datadog.go b/datadog/datadog.go index 27ef58644..d4cf1df65 100644 --- a/datadog/datadog.go +++ b/datadog/datadog.go @@ -70,6 +70,14 @@ func (c *StatsClient) Count(name string, value int64) { } } +// Count tracks the number of times something occurs per second with custom tags. +func (c *StatsClient) CountWithCustomTags(name string, value int64, t []string) { + tags := append(c.tags, t...) + if err := c.client.Count(name, value, tags, Rate); err != nil { + c.logger().Printf("datadog.StatsClient.Count error: %s", err) + } +} + // Gauge sets the value of a metric. func (c *StatsClient) Gauge(name string, value float64) { if err := c.client.Gauge(name, value, c.tags, Rate); err != nil { diff --git a/db.go b/db.go index 8bf33300e..8c176bb79 100644 --- a/db.go +++ b/db.go @@ -22,31 +22,31 @@ const ( // DB represents a container for frames. type DB struct { - mu sync.Mutex - path string - name string + mu sync.Mutex + path string + name string // Default time quantum for all frames in database. // This can be overridden by individual frames. - timeQuantum TimeQuantum + timeQuantum TimeQuantum // Label used for referring to columns in database. - columnLabel string + columnLabel string // Frames by name. - frames map[string]*Frame + frames map[string]*Frame // Max Slice on any node in the cluster, according to this node remoteMaxSlice uint64 remoteMaxInverseSlice uint64 // Profile attribute storage and cache - profileAttrStore *AttrStore + profileAttrStore *AttrStore - broadcaster Broadcaster - stats StatsClient + broadcaster Broadcaster + Stats StatsClient - LogOutput io.Writer + LogOutput io.Writer } // NewDB returns a new instance of DB. @@ -68,7 +68,7 @@ func NewDB(path, name string) (*DB, error) { columnLabel: DefaultColumnLabel, - stats: NopStatsClient, + Stats: NopStatsClient, LogOutput: ioutil.Discard, }, nil } @@ -165,7 +165,7 @@ func (db *DB) openFrames() error { } db.frames[fr.Name()] = fr - db.stats.Count("frameN", 1) + db.Stats.Count("frameN", 1) } return nil } @@ -405,7 +405,7 @@ func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) { // Add to database's frame lookup. db.frames[name] = f - db.stats.Count("frameN", 1) + db.Stats.Count("frameN", 1) return f, nil } @@ -416,7 +416,7 @@ func (db *DB) newFrame(path, name string) (*Frame, error) { return nil, err } f.LogOutput = db.LogOutput - f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name)) + f.Stats = db.Stats.WithTags(fmt.Sprintf("frame:%s", name)) f.broadcaster = db.broadcaster return f, nil } @@ -445,7 +445,7 @@ func (db *DB) DeleteFrame(name string) error { // Remove reference. delete(db.frames, name) - db.stats.Count("frameN", -1) + db.Stats.Count("frameN", -1) return nil } diff --git a/executor.go b/executor.go index 9c156ec56..52e4c7402 100644 --- a/executor.go +++ b/executor.go @@ -93,12 +93,13 @@ func (e *Executor) executeCall(ctx context.Context, db string, c *pql.Call, slic if err := e.validateCallArgs(c); err != nil { return nil, err } - + dbTag := fmt.Sprintf("db:%s", db) // Special handling for mutation and top-n calls. switch c.Name { case "ClearBit": return e.executeClearBit(ctx, db, c, opt) case "Count": + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeCount(ctx, db, c, slices, opt) case "SetBit": return e.executeSetBit(ctx, db, c, opt) @@ -107,8 +108,10 @@ func (e *Executor) executeCall(ctx context.Context, db string, c *pql.Call, slic case "SetProfileAttrs": return nil, e.executeSetProfileAttrs(ctx, db, c, opt) case "TopN": + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeTopN(ctx, db, c, slices, opt) default: + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeBitmapCall(ctx, db, c, slices, opt) } } @@ -485,6 +488,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Call } bm = bm.Union(f.Bitmap(rowID)) } + f.Stats.Count("range", 1) return bm, nil } @@ -770,6 +774,7 @@ func (e *Executor) executeSetBitmapAttrs(ctx context.Context, db string, c *pql. if err := frame.BitmapAttrStore().SetAttrs(rowID, attrs); err != nil { return err } + frame.Stats.Count("SetBitmapAttrs", 1) // Do not forward call if this is already being forwarded. if opt.Remote { @@ -855,6 +860,7 @@ func (e *Executor) executeBulkSetBitmapAttrs(ctx context.Context, db string, cal if err := frame.BitmapAttrStore().SetBulkAttrs(frameMap); err != nil { return nil, err } + frame.Stats.Count("SetBitmapAttrs", 1) } // Do not forward call if this is already being forwarded. @@ -914,7 +920,7 @@ func (e *Executor) executeSetProfileAttrs(ctx context.Context, db string, c *pql if err := d.ProfileAttrStore().SetAttrs(id, attrs); err != nil { return err } - + d.Stats.Count("SetProfileAttrs", 1) // Do not forward call if this is already being forwarded. if opt.Remote { return nil diff --git a/fragment.go b/fragment.go index 9bee4288e..7c6cff24d 100644 --- a/fragment.go +++ b/fragment.go @@ -938,7 +938,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { if err != nil { return err } - + f.stats.Count("ImportBit", 1) // import optimization to avoid linear foreach calls // slight risk of concurrent cache counter being off but // no real danger diff --git a/frame.go b/frame.go index 684b2225d..63c06f7c1 100644 --- a/frame.go +++ b/frame.go @@ -27,29 +27,29 @@ const ( // Frame represents a container for views. type Frame struct { - mu sync.Mutex - path string - db string - name string - timeQuantum TimeQuantum + mu sync.Mutex + path string + db string + name string + timeQuantum TimeQuantum - views map[string]*View + views map[string]*View // Bitmap attribute storage and cache bitmapAttrStore *AttrStore - broadcaster Broadcaster - stats StatsClient + broadcaster Broadcaster + Stats StatsClient // Frame settings. - rowLabel string - cacheType string - inverseEnabled bool + rowLabel string + cacheType string + inverseEnabled bool // Cache size for ranked frames - cacheSize uint32 + cacheSize uint32 - LogOutput io.Writer + LogOutput io.Writer } // NewFrame returns a new instance of frame. @@ -67,7 +67,7 @@ func NewFrame(path, db, name string) (*Frame, error) { views: make(map[string]*View), bitmapAttrStore: NewAttrStore(filepath.Join(path, ".data")), - stats: NopStatsClient, + Stats: NopStatsClient, rowLabel: DefaultRowLabel, inverseEnabled: DefaultInverseEnabled, @@ -256,7 +256,7 @@ func (f *Frame) openViews() error { view.BitmapAttrStore = f.bitmapAttrStore f.views[view.Name()] = view - f.stats.Count("maxSlice", 1) + f.Stats.Count("maxSlice", 1) } return nil @@ -422,7 +422,7 @@ func (f *Frame) newView(path, name string) *View { view.cacheType = f.cacheType view.LogOutput = f.LogOutput view.BitmapAttrStore = f.bitmapAttrStore - view.stats = f.stats.WithTags(fmt.Sprintf("slice:%s", name)) + view.stats = f.Stats.WithTags(fmt.Sprintf("slice:%s", name)) return view } diff --git a/index.go b/index.go index eb8f4b204..965c25ff0 100644 --- a/index.go +++ b/index.go @@ -245,7 +245,7 @@ func (i *Index) newDB(path, name string) (*DB, error) { return nil, err } db.LogOutput = i.LogOutput - db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) + db.Stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) db.broadcaster = i.Broadcaster return db, nil } diff --git a/stats.go b/stats.go index 1a6efe743..07c11a2f3 100644 --- a/stats.go +++ b/stats.go @@ -26,6 +26,9 @@ type StatsClient interface { // Tracks the number of times something occurs per second. Count(name string, value int64) + // Tracks the number of times something occurs per second with custom tags + CountWithCustomTags(name string, value int64, tags []string) + // Sets the value of a metric. Gauge(name string, value float64) @@ -47,6 +50,7 @@ type nopStatsClient struct{} func (c *nopStatsClient) Tags() []string { return nil } func (c *nopStatsClient) WithTags(tags ...string) StatsClient { return c } func (c *nopStatsClient) Count(name string, value int64) {} +func (c *nopStatsClient) CountWithCustomTags(name string, value int64, tags []string) {} func (c *nopStatsClient) Gauge(name string, value float64) {} func (c *nopStatsClient) Histogram(name string, value float64) {} func (c *nopStatsClient) Set(name string, value string) {} @@ -89,6 +93,10 @@ func (c *ExpvarStatsClient) Count(name string, value int64) { c.m.Add(name, value) } +func (c *ExpvarStatsClient) CountWithCustomTags(name string, value int64, tags []string) { + c.m.Add(name, value) +} + // Gauge sets the value of a metric. func (c *ExpvarStatsClient) Gauge(name string, value float64) { var f expvar.Float @@ -142,6 +150,14 @@ func (a MultiStatsClient) Count(name string, value int64) { } } + +func (a MultiStatsClient) CountWithCustomTags(name string, value int64, tags []string) { + for _, c := range a { + c.CountWithCustomTags(name, value, tags) + } +} + + // Gauge sets the value of a metric on all clients. func (a MultiStatsClient) Gauge(name string, value float64) { for _, c := range a { diff --git a/stats_test.go b/stats_test.go new file mode 100644 index 000000000..248b68ea0 --- /dev/null +++ b/stats_test.go @@ -0,0 +1,163 @@ +package pilosa_test + +import ( + "context" + "github.com/pilosa/pilosa" + "testing" + "time" +) + +func TestStatsCount_TopN(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth+2) + + // Execute query. + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + e.Index.Stats = &MockStats{ + mockCountWithTags: func(name string, value int64, tags []string) { + if name != "TopN" { + t.Errorf("Expected TopN, Results %s", name) + } + + if tags[0] != "db:d" { + t.Errorf("Expected db, Results %s", tags[0]) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=2)`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("CountWithCustomTags name isn't called") + } +} + +func TestStatsCount_Bitmap(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1) + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + e.Index.Stats = &MockStats{ + mockCountWithTags: func(name string, value int64, tags []string) { + if name != "Bitmap" { + t.Errorf("Expected Bitmap, Results %s", name) + } + + if tags[0] != "db:d" { + t.Errorf("Expected db, Results %s", tags[0]) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`Bitmap(frame=f, id=0)`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("CountWithCustomTags name isn't called") + } +} + +func TestStatsCount_SetBitmapAttrs(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 1) + + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + frame := e.Index.Frame("d", "f") + if frame == nil { + t.Fatal("frame not found") + } + + frame.Stats = &MockStats{ + mockCount: func(name string, value int64) { + if name != "SetBitmapAttrs" { + t.Errorf("Expected SetBitmapAttrs, Results %s", name) + } + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=10, frame=f, foo="bar")`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("Count isn't called") + } +} + +func TestStatsCount_SetProfileAttrs(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 1) + + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + db := e.Index.DB("d") + if db == nil { + t.Fatal("db not found") + } + + db.Stats = &MockStats{ + mockCount: func(name string, value int64) { + if name != "SetProfileAttrs" { + t.Errorf("Expected SetProfilepAttrs, Results %s", name) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`SetProfileAttrs(id=10, frame=f, foo="bar")`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("Count isn't called") + } +} + +type MockStats struct { + mockCount func(name string, value int64) + mockCountWithTags func(name string, value int64, tags []string) +} + +func (s *MockStats) Count(name string, value int64) { + if s.mockCount != nil { + s.mockCount(name, value) + return + } + return +} + +func (s *MockStats) CountWithCustomTags(name string, value int64, tags []string) { + if s.mockCountWithTags != nil { + s.mockCountWithTags(name, value, tags) + return + } + return +} + +func (c *MockStats) Tags() []string { return nil } +func (c *MockStats) WithTags(tags ...string) pilosa.StatsClient { return c } +func (c *MockStats) Gauge(name string, value float64) {} +func (c *MockStats) Histogram(name string, value float64) {} +func (c *MockStats) Set(name string, value string) {} +func (c *MockStats) Timing(name string, value time.Duration) {} \ No newline at end of file From 100fcfdc6ba0b42f372395c016d44067631dd988 Mon Sep 17 00:00:00 2001 From: Linh Vo Date: Thu, 20 Apr 2017 21:39:47 -0500 Subject: [PATCH 19/19] reformat --- db.go | 20 ++++++++++---------- frame.go | 26 +++++++++++++------------- stats.go | 16 +++++++--------- stats_test.go | 2 +- 4 files changed, 31 insertions(+), 33 deletions(-) diff --git a/db.go b/db.go index 8c176bb79..3cfbf23e1 100644 --- a/db.go +++ b/db.go @@ -22,31 +22,31 @@ const ( // DB represents a container for frames. type DB struct { - mu sync.Mutex - path string - name string + mu sync.Mutex + path string + name string // Default time quantum for all frames in database. // This can be overridden by individual frames. - timeQuantum TimeQuantum + timeQuantum TimeQuantum // Label used for referring to columns in database. - columnLabel string + columnLabel string // Frames by name. - frames map[string]*Frame + frames map[string]*Frame // Max Slice on any node in the cluster, according to this node remoteMaxSlice uint64 remoteMaxInverseSlice uint64 // Profile attribute storage and cache - profileAttrStore *AttrStore + profileAttrStore *AttrStore - broadcaster Broadcaster - Stats StatsClient + broadcaster Broadcaster + Stats StatsClient - LogOutput io.Writer + LogOutput io.Writer } // NewDB returns a new instance of DB. diff --git a/frame.go b/frame.go index 63c06f7c1..d173d9a07 100644 --- a/frame.go +++ b/frame.go @@ -27,29 +27,29 @@ const ( // Frame represents a container for views. type Frame struct { - mu sync.Mutex - path string - db string - name string - timeQuantum TimeQuantum + mu sync.Mutex + path string + db string + name string + timeQuantum TimeQuantum - views map[string]*View + views map[string]*View // Bitmap attribute storage and cache bitmapAttrStore *AttrStore - broadcaster Broadcaster - Stats StatsClient + broadcaster Broadcaster + Stats StatsClient // Frame settings. - rowLabel string - cacheType string - inverseEnabled bool + rowLabel string + cacheType string + inverseEnabled bool // Cache size for ranked frames - cacheSize uint32 + cacheSize uint32 - LogOutput io.Writer + LogOutput io.Writer } // NewFrame returns a new instance of frame. diff --git a/stats.go b/stats.go index 07c11a2f3..057d46320 100644 --- a/stats.go +++ b/stats.go @@ -47,14 +47,14 @@ var NopStatsClient StatsClient // nopStatsClient represents a client that doesn't do anything. type nopStatsClient struct{} -func (c *nopStatsClient) Tags() []string { return nil } -func (c *nopStatsClient) WithTags(tags ...string) StatsClient { return c } -func (c *nopStatsClient) Count(name string, value int64) {} +func (c *nopStatsClient) Tags() []string { return nil } +func (c *nopStatsClient) WithTags(tags ...string) StatsClient { return c } +func (c *nopStatsClient) Count(name string, value int64) {} func (c *nopStatsClient) CountWithCustomTags(name string, value int64, tags []string) {} -func (c *nopStatsClient) Gauge(name string, value float64) {} -func (c *nopStatsClient) Histogram(name string, value float64) {} -func (c *nopStatsClient) Set(name string, value string) {} -func (c *nopStatsClient) Timing(name string, value time.Duration) {} +func (c *nopStatsClient) Gauge(name string, value float64) {} +func (c *nopStatsClient) Histogram(name string, value float64) {} +func (c *nopStatsClient) Set(name string, value string) {} +func (c *nopStatsClient) Timing(name string, value time.Duration) {} // ExpvarStatsClient writes stats out to expvars. type ExpvarStatsClient struct { @@ -150,14 +150,12 @@ func (a MultiStatsClient) Count(name string, value int64) { } } - func (a MultiStatsClient) CountWithCustomTags(name string, value int64, tags []string) { for _, c := range a { c.CountWithCustomTags(name, value, tags) } } - // Gauge sets the value of a metric on all clients. func (a MultiStatsClient) Gauge(name string, value float64) { for _, c := range a { diff --git a/stats_test.go b/stats_test.go index 248b68ea0..cf06f2f06 100644 --- a/stats_test.go +++ b/stats_test.go @@ -160,4 +160,4 @@ func (c *MockStats) WithTags(tags ...string) pilosa.StatsClient { return c } func (c *MockStats) Gauge(name string, value float64) {} func (c *MockStats) Histogram(name string, value float64) {} func (c *MockStats) Set(name string, value string) {} -func (c *MockStats) Timing(name string, value time.Duration) {} \ No newline at end of file +func (c *MockStats) Timing(name string, value time.Duration) {}