diff --git a/Dockerfile b/Dockerfile index afbda01f2..529ffdff1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,7 +4,7 @@ COPY . pilosa RUN cd pilosa && CGO_ENABLED=0 make install FLAGS="-a" -FROM alpine:3.8 +FROM alpine:3.9.4 LABEL maintainer "dev@pilosa.com" diff --git a/api.go b/api.go index bdaa35b03..6a802526b 100644 --- a/api.go +++ b/api.go @@ -418,7 +418,7 @@ func (api *API) DeleteAvailableShard(_ context.Context, indexName, fieldName str api.server.logger.Printf("problem sending DeleteAvailableShard message: %s", err) return errors.Wrap(err, "sending DeleteAvailableShard message") } - api.holder.Stats.CountWithCustomTags("deleteAvailableShard", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName), fmt.Sprintf("field:%s", fieldName)}) + api.holder.Stats.CountWithCustomTags("deleteAvailableShard", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) return nil } diff --git a/cluster.go b/cluster.go index 674a0a267..cf0090c0f 100644 --- a/cluster.go +++ b/cluster.go @@ -21,6 +21,8 @@ import ( "hash/fnv" "io/ioutil" "math/rand" + "net/http" + "net/url" "os" "path/filepath" "sort" @@ -59,6 +61,10 @@ const ( resizeJobActionAdd = "ADD" resizeJobActionRemove = "REMOVE" + + confirmDownRetries = 10 + confirmDownSleep = 1 + confirmDownTimeout = 2 ) // Node represents a node in the cluster. @@ -1687,13 +1693,42 @@ func (c *cluster) considerTopology() error { return nil } +// band aid to protect against false nodeLeave events from memberlist +// the test is the lightest weight endpoint of the node in question /version +// TODO provide more robust solution to false nodeLeave events +func confirmNodeDown(uri URI, log logger.Logger) bool { + u := url.URL{ + Scheme: uri.Scheme, + Host: uri.HostPort(), + Path: "version", + } + ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second) + defer cancel() + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + log.Printf("bad request:%s %s", u.String(), err) + return false + } + + for i := 0; i < confirmDownRetries; i++ { + resp, err := http.DefaultClient.Do(req.WithContext(ctx)) + if err == nil { + if resp.StatusCode == 200 { + return false + } + } + log.Printf("NodeLeave Timeout with %s %d", uri.HostPort(), i) + time.Sleep(confirmDownSleep * time.Second) + } + return true +} + // ReceiveEvent represents an implementation of EventHandler. func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { // Ignore events sent from this node. if e.Node.ID == c.Node.ID { return nil } - switch e.Event { case NodeJoin: c.logger.Debugf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) @@ -1711,11 +1746,15 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { // not already removed by a removeNode request. We treat this as the // host being temporarily unavailable, and expect it to come back // up. - if c.removeNodeBasicSorted(e.Node.ID) { - c.Topology.nodeStates[e.Node.ID] = nodeStateDown - // put the cluster into STARTING if we've lost a number of nodes - // equal to or greater than ReplicaN - err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) + if confirmNodeDown(e.Node.URI, c.logger) { + if c.removeNodeBasicSorted(e.Node.ID) { + c.Topology.nodeStates[e.Node.ID] = nodeStateDown + // put the cluster into STARTING if we've lost a number of nodes + // equal to or greater than ReplicaN + err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) + } + } else { + c.logger.Printf("ignored received node leave: %v", e.Node) } } case NodeUpdate: diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 2a557694d..a30d98271 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -16,15 +16,24 @@ package pilosa import ( "bytes" + "fmt" "io/ioutil" "math/rand" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" "reflect" + "strconv" "strings" "testing" "testing/quick" "time" "github.com/davecgh/go-spew/spew" + "github.com/gorilla/mux" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" "github.com/pkg/errors" ) @@ -910,3 +919,71 @@ func TestCluster_UpdateCoordinator(t *testing.T) { } }) } + +func TestCluster_confirmNodeDownUp(t *testing.T) { + r := mux.NewRouter() + r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + fmt.Fprintln(w, "ignored") + })) + server := httptest.NewServer(r) + // Close the server when test finishes + defer server.Close() + u, err := url.Parse(server.URL) + if err != nil { + t.Error("bad test setup") + } + uri := URI{} + host, port, _ := net.SplitHostPort(u.Host) + uri.Scheme = u.Scheme + uri.Host = host + iport, err := strconv.ParseUint(port, 0, 16) + if err != nil { + t.Error(err) + } + uri.Port = uint16(iport) + if confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be up") + } + +} +func TestCluster_confirmNodeDownTimeout(t *testing.T) { + r := mux.NewRouter() + r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(confirmDownSleep * time.Second * confirmDownRetries) + fmt.Fprintln(w, "ignored") + })) + server := httptest.NewServer(r) + // Close the server when test finishes + defer server.Close() + u, err := url.Parse(server.URL) + if err != nil { + t.Error("bad test setup") + } + uri := URI{} + host, port, _ := net.SplitHostPort(u.Host) + uri.Scheme = u.Scheme + uri.Host = host + iport, err := strconv.ParseUint(port, 0, 16) + if err != nil { + t.Error(err) + } + uri.Port = uint16(iport) + + if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be down") + } + +} + +func TestCluster_confirmNodeDownDown(t *testing.T) { + uri := URI{} + uri.Scheme = "http" + uri.Host = "DoesntMatter" + uri.Port = 6666 + + if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be down") + } + +} diff --git a/docs/administration.md b/docs/administration.md index 33fa73104..6aa52fa2f 100644 --- a/docs/administration.md +++ b/docs/administration.md @@ -293,7 +293,7 @@ You can opt-out of the Pilosa diagnostics reporting by setting the command line ### Metrics -Pilosa can be configured to emit metrics pertaining to its internal processes in one of two formats: Expvar or StatsD. Metric recording is disabled by default. +Pilosa can be configured to emit metrics pertaining to its internal processes in one of three formats: Expvar, StatsD, or Prometheus. Metric recording is disabled by default. The metrics configuration options are: - [Host](../configuration/#metric-host): specify host that receives metric events diff --git a/docs/configuration.md b/docs/configuration.md index 05aabf970..8eb410de4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -293,7 +293,7 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h ``` #### Metric Service -* Description: Which stats service to use for collecting [metrics](../administration/#metrics). Choose from [statsd, expvar, none]. +* Description: Which stats service to use for collecting [metrics](../administration/#metrics). Choose from [statsd, expvar, prometheus, none]. * Flag: `--metric.service=statsd` * Env: `PILOSA_METRIC_SERVICE=statsd` * Config: diff --git a/field.go b/field.go index 3e71af589..3dd362ec0 100644 --- a/field.go +++ b/field.go @@ -300,12 +300,14 @@ func (f *Field) saveAvailableShards() error { } func (f *Field) unprotectedSaveAvailableShards() error { - // Open or create file. path := filepath.Join(f.path, ".available.shards") + // Create a temporary file to save to. + tempPath := path + tempExt - file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666) + // Open or create file. + file, err := os.OpenFile(tempPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666) if err != nil { - return errors.Wrap(err, "opening available shards file") + return errors.Wrap(err, "opening temporary available shards file") } defer file.Close() @@ -316,6 +318,11 @@ func (f *Field) unprotectedSaveAvailableShards() error { } bw.Flush() + // Move snapshot to data file location. + if err := os.Rename(tempPath, path); err != nil { + return fmt.Errorf("rename snapshot: %s", err) + } + return nil } @@ -514,6 +521,10 @@ func (f *Field) loadMeta() error { // saveMeta writes meta data for the field. func (f *Field) saveMeta() error { + path := filepath.Join(f.path, ".meta") + // Create a temporary file to marshal to. + tempPath := f.path + tempExt + // Marshal metadata. fo := f.options buf, err := proto.Marshal(fo.encode()) @@ -522,10 +533,15 @@ func (f *Field) saveMeta() error { } // Write to meta file. - if err := ioutil.WriteFile(filepath.Join(f.path, ".meta"), buf, 0666); err != nil { + if err := ioutil.WriteFile(tempPath, buf, 0666); err != nil { return errors.Wrap(err, "writing meta") } + // Move temp file to data file location. + if err := os.Rename(tempPath, path); err != nil { + return fmt.Errorf("rename temp: %s", err) + } + return nil } @@ -820,7 +836,7 @@ func (f *Field) newView(path, name string) *view { view := newView(path, f.index, f.name, name, f.options) view.logger = f.logger view.rowAttrStore = f.rowAttrStore - view.stats = f.Stats.WithTags(fmt.Sprintf("view:%s", name)) + view.stats = f.Stats view.broadcaster = f.broadcaster return view } diff --git a/fragment.go b/fragment.go index eaccf7b86..daa621ca5 100644 --- a/fragment.go +++ b/fragment.go @@ -74,6 +74,9 @@ const ( // cacheExt is the file extension for persisted cache ids. cacheExt = ".cache" + // tempExt is the file extension for temporary files. + tempExt = ".temp" + // HashBlockSize is the number of rows in a merkle hash block. HashBlockSize = 100 diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 19dc3f587..e45d3cce8 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -26,6 +26,7 @@ import ( "os" "reflect" "sort" + "sync/atomic" "testing" "testing/quick" @@ -104,6 +105,44 @@ func TestFragment_ClearBit(t *testing.T) { } } +// What about rowcache timing. +func TestFragment_RowcacheMap(t *testing.T) { + var done int64 + f := mustOpenFragment("i", "f", viewStandard, 0, "") + defer f.Clean(t) + + ch := make(chan struct{}) + + for i := 0; i < f.MaxOpN; i++ { + _, _ = f.setBit(0, uint64(i*32)) + } + // force snapshot so we get a mmapped row... + _ = f.snapshot() + row := f.row(0) + segment := row.Segments()[0] + bitmap := segment.data + + // request information from the frozen bitmap we got back + go func() { + for atomic.LoadInt64(&done) == 0 { + for i := 0; i < f.MaxOpN; i++ { + _ = bitmap.Contains(uint64(i * 32)) + } + } + close(ch) + }() + + // modify the original bitmap, until it causes a snapshot, which + // then invalidates the other map... + for j := 0; j < 5; j++ { + for i := 0; i < f.MaxOpN; i++ { + _, _ = f.setBit(0, uint64(i*32+j+1)) + } + } + atomic.StoreInt64(&done, 1) + <-ch +} + // Ensure a fragment can clear a row. func TestFragment_ClearRow(t *testing.T) { f := mustOpenFragment("i", "f", viewStandard, 0, "") diff --git a/go.mod b/go.mod index 397d0e267..883b3ca9e 100644 --- a/go.mod +++ b/go.mod @@ -13,15 +13,17 @@ require ( github.com/davecgh/go-spew v1.1.1 github.com/go-ole/go-ole v1.2.4 // indirect github.com/gogo/protobuf v1.2.0 - github.com/golang/protobuf v1.2.0 + github.com/golang/protobuf v1.3.1 github.com/google/go-cmp v0.2.0 github.com/gorilla/handlers v1.3.0 github.com/gorilla/mux v1.7.0 github.com/hashicorp/memberlist v0.1.3 github.com/inconshreveable/mousetrap v1.0.0 // indirect - github.com/opentracing/opentracing-go v1.0.2 + github.com/opentracing/opentracing-go v1.1.0 github.com/pelletier/go-toml v1.2.0 github.com/pkg/errors v0.8.1 + github.com/prometheus/client_golang v0.9.3 + github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 github.com/remyoudompheng/bigfft v0.0.0-20190321074620-2f0d2b0e0001 // indirect github.com/satori/go.uuid v1.2.0 github.com/shirou/gopsutil v2.18.12+incompatible @@ -29,14 +31,15 @@ require ( github.com/spf13/cobra v0.0.3 github.com/spf13/pflag v1.0.3 github.com/spf13/viper v1.3.1 + github.com/uber-go/atomic v1.4.0 // indirect github.com/uber/jaeger-client-go v2.16.0+incompatible github.com/uber/jaeger-lib v2.0.0+incompatible // indirect + go.uber.org/atomic v1.4.0 // indirect golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 // indirect golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect golang.org/x/sync v0.0.0-20190423024810-112230192c58 golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872 // indirect golang.org/x/text v0.3.2 // indirect - golang.org/x/tools v0.0.0-20190429231329-9d4d845e86f1 // indirect modernc.org/mathutil v1.0.0 modernc.org/strutil v1.0.0 ) diff --git a/go.sum b/go.sum index 674e2adea..5ed268a3e 100644 --- a/go.sum +++ b/go.sum @@ -8,9 +8,14 @@ github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/StackExchange/wmi v0.0.0-20181212234831-e0a55b97c705 h1:UUppSQnhf4Yc6xGxSkoQpPhb7RVzuv5Nb1mwJ5VId9s= github.com/StackExchange/wmi v0.0.0-20181212234831-e0a55b97c705/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg= +github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8= github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I= github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= +github.com/beorn7/perks v1.0.0 h1:HWo1m869IqiPhD389kmkxeTalrjNbbJTC8LXupb+sl0= +github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4= github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps= github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko= @@ -22,14 +27,22 @@ github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8Nz github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= +github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= +github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= +github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI= github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM= +github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.2.0 h1:xU6/SpYbvkNYiptHJYEDRseDLvYE7wSqhYYNy0QSUzI= github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg= +github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c h1:964Od4U6p2jUkFxvCydnIczKteheJEzHRToSGK3Bnlw= github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ= github.com/google/go-cmp v0.2.0 h1:+dTQ8DZQJz0Mb/HjFlkptS1FeQ4cWSnN941F8aEG4SQ= @@ -54,28 +67,47 @@ github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCO github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= -github.com/hashicorp/memberlist v0.1.3 h1:EmmoJme1matNzb+hMpDuR/0sbJSUisxyqBGG676r31M= -github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM= github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8= +github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= +github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY= github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ= +github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= +github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/miekg/dns v1.0.14 h1:9jZdLNd/P4+SfEJ0TNyxYpsK8N4GtfylBLqtbYN1sbA= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE= github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y= -github.com/opentracing/opentracing-go v1.0.2 h1:3jA2P6O1F9UOrWVpwrIo17pu01KWvNWg4X946/Y5Zwg= -github.com/opentracing/opentracing-go v1.0.2/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= +github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U= +github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU= +github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c h1:Lgl0gzECD8GnQ5QCWA8o6BtfL6mDH5rQgM4/fX3avOs= github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= +github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_golang v0.9.3 h1:9iH4JKXLzFbOAdtqv/a+j8aewx2Y8lAjAydhbaScPF8= +github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90 h1:S/YWwWx/RA8rT8tKFRuGUZhuA90OyIBpPCXkcbwU8DE= +github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro= +github.com/prometheus/common v0.4.0 h1:7etb9YClo3a6HjLzfl6rIQaU+FDfi0VSX39io3aQ+DM= +github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4= +github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 h1:sofwID9zm4tzrgykg80hfFph1mryUeLRsUfoocVVmRY= +github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= +github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU= github.com/remyoudompheng/bigfft v0.0.0-20190321074620-2f0d2b0e0001 h1:YDeskXpkNDhPdWN3REluVa46HQOVuVkjkd2sWnrABNQ= github.com/remyoudompheng/bigfft v0.0.0-20190321074620-2f0d2b0e0001/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= @@ -86,6 +118,8 @@ github.com/shirou/gopsutil v2.18.12+incompatible h1:1eaJvGomDnH74/5cF4CTmTbLHAri github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U= github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc= +github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= +github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI= github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ= @@ -99,22 +133,20 @@ github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg= github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= github.com/spf13/viper v1.3.1 h1:5+8j8FTpnFV4nEImW/ofkzEt8VoOiLXxdYIDsB73T38= github.com/spf13/viper v1.3.1/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s= +github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= -github.com/uber-go/atomic v1.3.2 h1:Azu9lPBWRNKzYXSIwRfgRuDuS0YKsK4NFhiQv98gkxo= -github.com/uber-go/atomic v1.3.2/go.mod h1:/Ct5t2lcmbJ4OSe/waGBoaVvVqtO0bmtfVNex1PFV8g= -github.com/uber/jaeger-client-go v2.15.0+incompatible h1:NP3qsSqNxh8VYr956ur1N/1C1PjvOJnJykCzcD5QHbk= -github.com/uber/jaeger-client-go v2.15.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk= +github.com/uber-go/atomic v1.4.0 h1:yOuPqEq4ovnhEjpHmfFwsqBXDYbQeT6Nb0bwD6XnD5o= +github.com/uber-go/atomic v1.4.0/go.mod h1:/Ct5t2lcmbJ4OSe/waGBoaVvVqtO0bmtfVNex1PFV8g= github.com/uber/jaeger-client-go v2.16.0+incompatible h1:Q2Pp6v3QYiocMxomCaJuwQGFt7E53bPYqEgug/AoBtY= github.com/uber/jaeger-client-go v2.16.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk= -github.com/uber/jaeger-lib v1.5.0 h1:OHbgr8l656Ub3Fw5k9SWnBfIEwvoHQ+W2y+Aa9D1Uyo= -github.com/uber/jaeger-lib v1.5.0/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U= github.com/uber/jaeger-lib v2.0.0+incompatible h1:iMSCV0rmXEogjNWPh2D0xk9YVKvrtGoHJNe9ebLu/pw= github.com/uber/jaeger-lib v2.0.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U= github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0= github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q= -go.uber.org/atomic v1.3.2 h1:2Oa65PReHzfn29GpvgsYwloV9AVFHPDk8tYxt2c2tr4= -go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= +go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU= +go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= +golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9 h1:mKdxBk7AujPs8kU4m80U72y/zjbZ3UcXC7dClwKbUI0= golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= @@ -123,15 +155,19 @@ golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 h1:p/H982KKEjUnLJkM3tt/Le golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519 h1:x6rhz8Y9CjbgQkccRGmELH6K+LJj7tOoh3XWeC1yaQM= golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= -golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 h1:FP8hkuE6yUEaJnK7O2eTuejKWwW+Rhfj80dQ2JcKxCU= golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a h1:1n5lsVfiQW3yfsRGu98756EH1YthsFqr/5mxHduZW2A= golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -143,10 +179,10 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= -golang.org/x/tools v0.0.0-20190429231329-9d4d845e86f1 h1:MSSXVSCgrxTAYytvleklMKlLdxjexiJWNffJciO1nCI= -golang.org/x/tools v0.0.0-20190429231329-9d4d845e86f1/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q= +gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= modernc.org/mathutil v1.0.0 h1:93vKjrJopTPrtTNpZ8XIovER7iCIH1QU7wNbOQXC60I= diff --git a/http/handler.go b/http/handler.go index 8e932271a..d1ea07ae9 100644 --- a/http/handler.go +++ b/http/handler.go @@ -39,6 +39,7 @@ import ( "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/tracing" "github.com/pkg/errors" + "github.com/prometheus/client_golang/prometheus/promhttp" ) // Handler represents an HTTP handler. @@ -234,6 +235,41 @@ func (h *Handler) extractTracing(next http.Handler) http.Handler { }) } +func (h *Handler) collectStats(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t := time.Now() + next.ServeHTTP(w, r) + dur := time.Since(t) + + statsTags := make([]string, 0, 5) + + longQueryTime := h.api.LongQueryTime() + if longQueryTime > 0 && dur > longQueryTime { + h.logger.Printf("%s %s %v", r.Method, r.URL.String(), dur) + statsTags = append(statsTags, "slow_query") + } + + pathParts := strings.Split(r.URL.Path, "/") + if externalPrefixFlag[pathParts[1]] { + statsTags = append(statsTags, "external") + } + + statsTags = append(statsTags, "useragent:"+r.UserAgent()) + + path, err := mux.CurrentRoute(r).GetPathTemplate() + if err == nil { + statsTags = append(statsTags, "path:"+path) + } + + statsTags = append(statsTags, "method:"+r.Method) + + stats := h.api.StatsWithTags(statsTags) + if stats != nil { + stats.Timing("http.request", dur, 0.1) + } + }) +} + // newRouter creates a new mux http router. func newRouter(handler *Handler) *mux.Router { router := mux.NewRouter() @@ -243,6 +279,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/cluster/resize/set-coordinator", handler.handlePostClusterResizeSetCoordinator).Methods("POST").Name("PostClusterResizeSetCoordinator") router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET") router.Handle("/debug/vars", expvar.Handler()).Methods("GET") + router.Handle("/metrics", promhttp.Handler()) router.HandleFunc("/export", handler.handleGetExport).Methods("GET").Name("GetExport") router.HandleFunc("/index", handler.handleGetIndexes).Methods("GET").Name("GetIndexes") router.HandleFunc("/index/{index}", handler.handleGetIndex).Methods("GET").Name("GetIndex") @@ -278,6 +315,7 @@ func newRouter(handler *Handler) *mux.Router { router.Use(handler.queryArgValidator) router.Use(handler.extractTracing) + router.Use(handler.collectStats) return router } @@ -293,32 +331,7 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } }() - t := time.Now() h.Handler.ServeHTTP(w, r) - dif := time.Since(t) - - // Calculate per request StatsD metrics when the handler is fully configured. - statsTags := make([]string, 0, 3) - - longQueryTime := h.api.LongQueryTime() - if longQueryTime > 0 && dif > longQueryTime { - h.logger.Printf("%s %s %v", r.Method, r.URL.String(), dif) - statsTags = append(statsTags, "slow_query") - } - - pathParts := strings.Split(r.URL.Path, "/") - endpointName := strings.Join(pathParts, "_") - - if externalPrefixFlag[pathParts[1]] { - statsTags = append(statsTags, "external") - } - - // useragent tag identifies internal/external endpoints - statsTags = append(statsTags, "useragent:"+r.UserAgent()) - stats := h.api.StatsWithTags(statsTags) - if stats != nil { - stats.Histogram("http."+endpointName, float64(dif), 0.1) - } } // successResponse is a general success/error struct for http responses. @@ -1485,10 +1498,6 @@ type defaultClusterMessageResponse struct{} // translateStoreBufferSize is the buffer size used for streaming data. const translateStoreBufferSize = 1 << 16 // 64k -// translateStoreBufferSizeMax is the maximum size that the buffer is allowed -// to grow before raising an error. -const translateStoreBufferSizeMax = 1 << 22 // 4Mb - func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() offset, _ := strconv.ParseInt(q.Get("offset"), 10, 64) @@ -1517,16 +1526,6 @@ func (h *Handler) handleGetTranslateData(w http.ResponseWriter, r *http.Request) n, err := rdr.Read(buf) if err == io.EOF { return - } else if err == pilosa.ErrTranslateReadTargetUndersized { - // Increase the buffer size and try to read again. - useBufferSize *= 2 - // Prevent the buffer from growing without bound. - if useBufferSize > translateStoreBufferSizeMax { - h.logger.Printf("http: translate store buffer exceeded max size: %s", err) - return - } - buf = make([]byte, useBufferSize) - continue } else if err != nil { h.logger.Printf("http: translate store read error: %s", err) return diff --git a/index.go b/index.go index dbb9c052a..bef4ff38d 100644 --- a/index.go +++ b/index.go @@ -405,7 +405,7 @@ func (i *Index) newField(path, name string) (*Field, error) { return nil, err } f.logger = i.logger - f.Stats = i.Stats.WithTags(fmt.Sprintf("field:%s", name)) + f.Stats = i.Stats f.broadcaster = i.broadcaster f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data")) return f, nil diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 612189c1b..5c8f4ed6b 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -80,7 +80,7 @@ func TestClusterStuff(t *testing.T) { // TODO change the sleep to wait for status to return to NORMAL - need support in internal client for getting status t.Log("done with pause, waiting for stability") - time.Sleep(time.Second * 3) + time.Sleep(time.Second * 20) t.Log("done waiting for stability") r, err = cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"}) diff --git a/prometheus/prometheus.go b/prometheus/prometheus.go new file mode 100644 index 000000000..b29267019 --- /dev/null +++ b/prometheus/prometheus.go @@ -0,0 +1,301 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package prometheus + +import ( + "sort" + "strings" + "sync" + "time" + + "github.com/pilosa/pilosa/logger" + "github.com/pilosa/pilosa/stats" + "github.com/prometheus/client_golang/prometheus" +) + +const ( + // namespace is prepended to each metric event name with "_" + namespace = "pilosa" +) + +// Ensure client implements interface. +var _ stats.StatsClient = &prometheusClient{} + +// Module-level mutex to avoid copying in WithTags() +var mu sync.Mutex + +// prometheusClient represents a Prometheus implementation of pilosa.statsClient. +type prometheusClient struct { + tags []string + logger logger.Logger + counters map[string]prometheus.Counter + counterVecs map[string]*prometheus.CounterVec + gauges map[string]prometheus.Gauge + gaugeVecs map[string]*prometheus.GaugeVec + observers map[string]prometheus.Observer + summaryVecs map[string]*prometheus.SummaryVec +} + +// NewPrometheusClient returns a new instance of StatsClient. +func NewPrometheusClient() (*prometheusClient, error) { + return &prometheusClient{ + logger: logger.NopLogger, + counters: make(map[string]prometheus.Counter), + counterVecs: make(map[string]*prometheus.CounterVec), + gauges: make(map[string]prometheus.Gauge), + gaugeVecs: make(map[string]*prometheus.GaugeVec), + observers: make(map[string]prometheus.Observer), + summaryVecs: make(map[string]*prometheus.SummaryVec), + }, nil +} + +// Open no-op to satisfy interface +func (c *prometheusClient) Open() {} + +// Close no-op to satisfy interface +func (c *prometheusClient) Close() error { + return nil +} + +// Tags returns a sorted list of tags on the client. +func (c *prometheusClient) Tags() []string { + return c.tags +} + +// labels returns an instance of prometheus.Labels with the value of the set tags. +func (c *prometheusClient) labels() prometheus.Labels { + return tagsToLabels(c.tags) +} + +// WithTags returns a new client with additional tags appended. +func (c *prometheusClient) WithTags(tags ...string) stats.StatsClient { + return &prometheusClient{ + tags: unionStringSlice(c.tags, tags), + logger: c.logger, + counters: c.counters, + counterVecs: c.counterVecs, + gauges: c.gauges, + gaugeVecs: c.gaugeVecs, + observers: c.observers, + summaryVecs: c.summaryVecs, + } +} + +// Count tracks the number of times something occurs per second. +func (c *prometheusClient) Count(name string, value int64, rate float64) { + mu.Lock() + defer mu.Unlock() + + var counter prometheus.Counter + var ok bool + name = strings.Replace(name, ".", "_", -1) + labels := c.labels() + opts := prometheus.CounterOpts{ + Namespace: namespace, + Name: name, + } + if len(labels) == 0 { + counter, ok = c.counters[name] + if !ok { + counter = prometheus.NewCounter(opts) + c.counters[name] = counter + prometheus.MustRegister(counter) + } + } else { + var counterVec *prometheus.CounterVec + counterVec, ok = c.counterVecs[name] + if !ok { + counterVec = prometheus.NewCounterVec( + opts, + labelKeys(labels), + ) + c.counterVecs[name] = counterVec + prometheus.MustRegister(counterVec) + } + var err error + counter, err = counterVec.GetMetricWith(labels) + if err != nil { + c.logger.Printf("counterVec.GetMetricWith error: %s", err) + } + } + if value == 1 { + counter.Inc() + } else { + counter.Add(float64(value)) + } +} + +// CountWithCustomTags tracks the number of times something occurs per second with custom tags. +func (c *prometheusClient) CountWithCustomTags(name string, value int64, rate float64, t []string) { + c.WithTags(append(c.tags, t...)...).Count(name, value, rate) +} + +// Gauge sets the value of a metric. +func (c *prometheusClient) Gauge(name string, value float64, rate float64) { + mu.Lock() + defer mu.Unlock() + + var gauge prometheus.Gauge + var ok bool + name = strings.Replace(name, ".", "_", -1) + labels := c.labels() + opts := prometheus.GaugeOpts{ + Namespace: namespace, + Name: name, + } + if len(labels) == 0 { + gauge, ok = c.gauges[name] + if !ok { + gauge = prometheus.NewGauge(opts) + c.gauges[name] = gauge + prometheus.MustRegister(gauge) + } + } else { + var gaugeVec *prometheus.GaugeVec + gaugeVec, ok = c.gaugeVecs[name] + if !ok { + gaugeVec = prometheus.NewGaugeVec( + opts, + labelKeys(labels), + ) + c.gaugeVecs[name] = gaugeVec + prometheus.MustRegister(gaugeVec) + } + var err error + gauge, err = gaugeVec.GetMetricWith(labels) + if err != nil { + c.logger.Printf("gaugeVec.GetMetricWith error: %s", err) + return + } + } + gauge.Set(float64(value)) +} + +// Histogram tracks statistical distribution of a metric. +func (c *prometheusClient) Histogram(name string, value float64, rate float64) { + mu.Lock() + defer mu.Unlock() + + var observer prometheus.Observer + var ok bool + name = strings.Replace(name, ".", "_", -1) + labels := c.labels() + opts := prometheus.SummaryOpts{ + Namespace: namespace, + Name: name, + Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + } + if len(labels) == 0 { + observer, ok = c.observers[name] + if !ok { + summary := prometheus.NewSummary(opts) + observer = summary + c.observers[name] = observer + prometheus.MustRegister(summary) + } + } else { + var summaryVec *prometheus.SummaryVec + summaryVec, ok = c.summaryVecs[name] + if !ok { + summaryVec = prometheus.NewSummaryVec( + opts, + labelKeys(labels), + ) + c.summaryVecs[name] = summaryVec + prometheus.MustRegister(summaryVec) + } + var err error + observer, err = summaryVec.GetMetricWith(labels) + if err != nil { + c.logger.Printf("summaryVec.GetMetricWith error: %s", err) + return + } + } + observer.Observe(value) +} + +// Set tracks number of unique elements. +func (c *prometheusClient) Set(name string, value string, rate float64) { + c.logger.Printf("prometheusClient.Set unimplemented: %s=%s", name, value) +} + +// Timing tracks timing information for a metric. +func (c *prometheusClient) Timing(name string, value time.Duration, rate float64) { + durationMs := value / time.Second + c.Histogram(name, float64(durationMs), rate) +} + +// SetLogger sets the logger for client. +func (c *prometheusClient) SetLogger(logger logger.Logger) { + c.logger = logger +} + +// unionStringSlice returns a sorted set of tags which combine a & b. +func unionStringSlice(a, b []string) []string { + // Sort both sets first. + sort.Strings(a) + sort.Strings(b) + + // Find size of largest slice. + n := len(a) + if len(b) > n { + n = len(b) + } + + // Exit if both sets are empty. + if n == 0 { + return nil + } + + // Iterate over both in order and merge. + other := make([]string, 0, n) + for len(a) > 0 || len(b) > 0 { + if len(a) == 0 { + other, b = append(other, b[0]), b[1:] + } else if len(b) == 0 { + other, a = append(other, a[0]), a[1:] + } else if a[0] < b[0] { + other, a = append(other, a[0]), a[1:] + } else if b[0] < a[0] { + other, b = append(other, b[0]), b[1:] + } else { + other, a, b = append(other, a[0]), a[1:], b[1:] + } + } + return other +} + +func tagsToLabels(tags []string) (labels prometheus.Labels) { + labels = make(prometheus.Labels) + for _, tag := range tags { + tagParts := strings.SplitAfterN(tag, ":", 2) + if len(tagParts) != 2 { + // only process tags in "key:value" form + continue + } + labels[tagParts[0][0:len(tagParts[0])-1]] = tagParts[1] + } + return labels +} + +func labelKeys(labels prometheus.Labels) (keys []string) { + keys = make([]string, len(labels)) + i := 0 + for k := range labels { + keys[i] = k + i++ + } + return keys +} diff --git a/prometheus/prometheus_test.go b/prometheus/prometheus_test.go new file mode 100644 index 000000000..d40a0ecd8 --- /dev/null +++ b/prometheus/prometheus_test.go @@ -0,0 +1,82 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package prometheus_test + +import ( + "reflect" + "testing" + "time" + + pilosaPrometheus "github.com/pilosa/pilosa/prometheus" + "github.com/prometheus/client_golang/prometheus" + io_prometheus_client "github.com/prometheus/client_model/go" +) + +func TestPrometheusClient_WithTags(t *testing.T) { + // Create a new client. + c, err := pilosaPrometheus.NewPrometheusClient() + if err != nil { + t.Fatal(err) + } + defer c.Close() + + // Create a new client with additional tags. + c1 := c.WithTags("foo", "bar") + if tags := c1.Tags(); !reflect.DeepEqual(tags, []string{"bar", "foo"}) { + t.Fatalf("unexpected tags: %+v", tags) + } + + // Create a new client from the clone with more tags. + c2 := c1.WithTags("bar", "baz") + if tags := c2.Tags(); !reflect.DeepEqual(tags, []string{"bar", "baz", "foo"}) { + t.Fatalf("unexpected tags: %+v", tags) + } +} + +func TestPrometheusClient_Methods(t *testing.T) { + // Create a new client. + c, err := pilosaPrometheus.NewPrometheusClient() + if err != nil { + t.Fatal(err) + } + defer c.Close() + + dur, _ := time.ParseDuration("123us") + c.CountWithCustomTags("ct", 1, 1.0, []string{"foo:bar"}) + c.Count("cc", 1, 1.0) + c.Gauge("gg", 10, 1.0) + c.Histogram("hh", 1, 1.0) + c.Timing("tt", dur, 1.0) + + metricFams, err := prometheus.DefaultGatherer.Gather() + if err != nil { + t.Fatal(err) + } + for _, metricName := range []string{"pilosa_ct", "pilosa_cc", "pilosa_gg", "pilosa_hh", "pilosa_tt"} { + if metricExists(metricName, metricFams) { + continue + } + t.Fatalf("Metric was not recorded: %s", metricName) + } +} + +func metricExists(metricName string, metricFams []*io_prometheus_client.MetricFamily) bool { + for _, metricFam := range metricFams { + if metricFam.GetName() == metricName { + return true + } + } + return false +} diff --git a/roaring/container_stash.go b/roaring/container_stash.go index 46725238f..fe03f9d2e 100644 --- a/roaring/container_stash.go +++ b/roaring/container_stash.go @@ -256,11 +256,20 @@ func (c *Container) setMapped(mapped bool) { } // Freeze returns an unmodifiable container identical to c. This might -// be c, now marked unmodifiable, or might be a new container. +// be c, now marked unmodifiable, or might be a new container. If c +// is currently marked as "mapped", referring to a backing store that's +// not a conventional Go pointer, the storage may be copied. func (c *Container) Freeze() *Container { if c == nil { return nil } + // don't need to freeze + if c.flags&flagFrozen != 0 { + return c + } + // unmapOrClone should unmap-in-place because the existing + // container isn't frozen (or we'd already have returned it). + c = c.unmapOrClone() c.flags |= flagFrozen return c } diff --git a/roaring/containers_slice.go b/roaring/containers_slice.go index cbff4f179..df44b4ff1 100644 --- a/roaring/containers_slice.go +++ b/roaring/containers_slice.go @@ -196,9 +196,7 @@ func (sc *sliceContainers) Update(key uint64, fn func(*Container, bool) (*Contai // don't expand the slice just to add a nil container, we // could return that anyway if write && nc != nil { - sc.containers = append(sc.containers, nil) - copy(sc.containers[i+1:], sc.containers[i:]) - sc.containers[i] = nc + sc.insertAt(key, nc, -i-1) } } } diff --git a/roaring/roaring.go b/roaring/roaring.go index 023c75bc4..31916445f 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -971,11 +971,9 @@ func (b *Bitmap) countEmptyContainers() int { // Optimize converts array and bitmap containers to run containers as necessary. func (b *Bitmap) Optimize() { - citer, _ := b.Containers.Iterator(0) - for citer.Next() { - _, c := citer.Value() - c.optimize() - } + b.Containers.UpdateEvery(func(c *Container, existed bool) (*Container, bool) { + return c.optimize(), true + }) } type errWriter struct { @@ -3519,7 +3517,7 @@ RUNLOOP: } } output := NewContainerRun(runs) - output.optimize() + output = output.optimize() return output } diff --git a/server/server.go b/server/server.go index e376c5aa6..743accf95 100644 --- a/server/server.go +++ b/server/server.go @@ -44,6 +44,7 @@ import ( "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/http" "github.com/pilosa/pilosa/logger" + "github.com/pilosa/pilosa/prometheus" "github.com/pilosa/pilosa/stats" "github.com/pilosa/pilosa/statsd" "github.com/pilosa/pilosa/syswrap" @@ -395,6 +396,8 @@ func newStatsClient(name string, host string) (stats.StatsClient, error) { return stats.NewExpvarStatsClient(), nil case "statsd": return statsd.NewStatsClient(host) + case "prometheus": + return prometheus.NewPrometheusClient() case "nop", "none": return stats.NopStatsClient, nil default: diff --git a/stats/stats_test.go b/stats/stats_test.go index f09cce79e..483befcb0 100644 --- a/stats/stats_test.go +++ b/stats/stats_test.go @@ -45,39 +45,39 @@ func TestMultiStatClient_Expvar(t *testing.T) { hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2) hldr.ClearBit("d", "f", 0, 1) - if stats.Expvar.String() != `{"index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + if stats.Expvar.String() != `{"index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } hldr.Stats.CountWithCustomTags("cc", 1, 1.0, []string{"foo:bar"}) - if stats.Expvar.String() != `{"cc": 1, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + if stats.Expvar.String() != `{"cc": 1, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Gauge creates a unique key, subsequent Gauge calls will overwrite hldr.Stats.Gauge("g", 5, 1.0) hldr.Stats.Gauge("g", 8, 1.0) - if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}}` { + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Set creates a unique key, subsequent sets will overwrite hldr.Stats.Set("s", "4", 1.0) hldr.Stats.Set("s", "7", 1.0) - if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7"}` { + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7"}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Record timing duration and a uniquely Set key/value dur, _ := time.ParseDuration("123us") hldr.Stats.Timing("tt", dur, 1.0) - if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { + if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7", "tt": 123µs}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } // Expvar histogram is implemented as a gauge hldr.Stats.Histogram("hh", 3, 1.0) - if stats.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": {"field:f": {"view:standard": {"shard:0": {"clearBit": 1, "rows": 0, "setBit": 2}, "shard:1": {"rows": 0, "setBit": 2}}}}, "s": "7", "tt": 123µs}` { + if stats.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7", "tt": 123µs}` { t.Fatalf("unexpected expvar : %s", stats.Expvar.String()) } diff --git a/translate.go b/translate.go index 94eb57186..1b6eb6244 100644 --- a/translate.go +++ b/translate.go @@ -45,11 +45,10 @@ const ( // Translate store errors. var ( - ErrTranslateStoreClosed = errors.New("pilosa: translate store closed") - ErrTranslateStoreReaderClosed = errors.New("pilosa: translate store reader closed") - ErrReplicationNotSupported = errors.New("pilosa: replication not supported") - ErrTranslateStoreReadOnly = errors.New("pilosa: translate store could not find or create key, translate store read only") - ErrTranslateReadTargetUndersized = errors.New("pilosa: translate read target is undersized") + ErrTranslateStoreClosed = errors.New("pilosa: translate store closed") + ErrTranslateStoreReaderClosed = errors.New("pilosa: translate store reader closed") + ErrReplicationNotSupported = errors.New("pilosa: replication not supported") + ErrTranslateStoreReadOnly = errors.New("pilosa: translate store could not find or create key, translate store read only") ) // TranslateStore is the storage for translation string-to-uint64 values. @@ -746,48 +745,38 @@ func (e *LogEntry) ReadFrom(r io.Reader) (_ int64, err error) { return int64(uVarintSize(e.Length)), err } - // Slurp entire entry and replace reader. - buf := make([]byte, e.Length) - n, err := io.ReadFull(r, buf) - n64 := int64(n + uVarintSize(e.Length)) - if err != nil { - return n64, err - } - bufr := bytes.NewReader(buf) - br, r = bufr, bufr - // Read the entry type. if err := binary.Read(r, binary.BigEndian, &e.Type); err != nil { - return n64, err + return 0, err } // Read index name. if sz, err := binary.ReadUvarint(br); err != nil { - return n64, err + return 0, err } else if sz == 0 { e.Index = nil } else { e.Index = make([]byte, sz) if _, err := io.ReadFull(r, e.Index); err != nil { - return n64, err + return 0, err } } // Read field name. if sz, err := binary.ReadUvarint(br); err != nil { - return n64, err + return 0, err } else if sz == 0 { e.Field = nil } else { e.Field = make([]byte, sz) if _, err := io.ReadFull(r, e.Field); err != nil { - return n64, err + return 0, err } } // Read key count. if n, err := binary.ReadUvarint(br); err != nil { - return n64, err + return 0, err } else if n == 0 { e.IDs, e.Keys = nil, nil } else { @@ -798,20 +787,21 @@ func (e *LogEntry) ReadFrom(r io.Reader) (_ int64, err error) { for i := range e.Keys { // Read identifier. if e.IDs[i], err = binary.ReadUvarint(br); err != nil { - return n64, err + return 0, err } // Read key. if sz, err := binary.ReadUvarint(br); err != nil { - return n64, err + return 0, err } else if sz > 0 { e.Keys[i] = make([]byte, sz) if _, err := io.ReadFull(r, e.Keys[i]); err != nil { - return n64, err + return 0, err } } } - return n64, nil + + return int64(uVarintSize(e.Length)) + int64(e.Length), nil } // WriteTo serializes a LogEntry to w. @@ -875,22 +865,6 @@ func (e *LogEntry) WriteTo(w io.Writer) (_ int64, err error) { return int64(sz) + n, err } -// validLogEntriesLen returns the maximum length of p that contains valid entries. -func validLogEntriesLen(p []byte) (n int) { - r := bytes.NewReader(p) - for { - if sz, err := binary.ReadUvarint(r); err != nil { - return n - } else if off, err := r.Seek(int64(sz), io.SeekCurrent); err != nil { - return n - } else if off > int64(len(p)) { - return n - } else { - n = int(off) - } - } -} - type fieldKey struct { index string field string @@ -1126,7 +1100,7 @@ func (r *translateFileReader) Read(p []byte) (n int, err error) { } } -// read writes the bytes for zero or more valid entries to p. +// read reads up to len(p) bytes into p. func (r *translateFileReader) read(p []byte) (n int, err error) { sz := r.store.size() @@ -1137,20 +1111,13 @@ func (r *translateFileReader) read(p []byte) (n int, err error) { return 0, nil } - if max := sz - r.offset; max > int64(len(p)) { - // If p is not large enough to hold a single entry, - // return an error so the client can increase the - // size of p and try again. - return 0, ErrTranslateReadTargetUndersized - } else if int64(len(p)) > max { + if max := sz - r.offset; int64(len(p)) > max { // Shorten buffer to maximum read size. p = p[:max] } // Read data from file at offset. - // Limit the number of bytes read to only whole entries. n, err = r.file.ReadAt(p, r.offset) - n = validLogEntriesLen(p[:n]) r.offset += int64(n) return n, err } diff --git a/translate_test.go b/translate_test.go index 4fa977e62..344a820d5 100644 --- a/translate_test.go +++ b/translate_test.go @@ -370,6 +370,45 @@ func TestTranslateFile_Reader(t *testing.T) { t.Fatal(diff) } }) + + t.Run("TinyBuffer", func(t *testing.T) { + stringKeys := make([]string, 1024) + byteSliceKeys := make([][]byte, len(stringKeys)) + ids := make([]uint64, len(stringKeys)) + for i := range stringKeys { + stringKeys[i] = fmt.Sprintf("KEY%d", i) + byteSliceKeys[i] = []byte(stringKeys[i]) + ids[i] = uint64(i + 1) + } + + s := MustOpenTranslateFile() + defer s.MustClose() + if _, err := s.TranslateColumnsToUint64("IDX0", stringKeys); err != nil { + t.Fatal(err) + } + + // Obtain the reader and use the smallest possible buffer for bufio. + rc, err := s.Reader(context.Background(), 0) + if err != nil { + t.Fatal(err) + } + brc := bufio.NewReaderSize(rc, 16) + defer rc.Close() + + // Record should be able to be read using multiple reads. + var entry pilosa.LogEntry + if _, err := entry.ReadFrom(brc); err != nil { + t.Fatal(err) + } else if diff := cmp.Diff(entry, pilosa.LogEntry{ + Type: pilosa.LogEntryTypeInsertColumn, + Index: []byte("IDX0"), + IDs: ids, + Keys: byteSliceKeys, + Length: 9012, + }); diff != "" { + t.Fatal(diff) + } + }) } func TestPrintTranslateFile(t *testing.T) { diff --git a/view.go b/view.go index eee1ab810..5aba4dbd3 100644 --- a/view.go +++ b/view.go @@ -267,7 +267,7 @@ func (v *view) newFragment(path string, shard uint64) *fragment { frag.CacheType = v.cacheType frag.CacheSize = v.cacheSize frag.Logger = v.logger - frag.stats = v.stats.WithTags(fmt.Sprintf("shard:%d", shard)) + frag.stats = v.stats if v.fieldType == FieldTypeMutex { frag.mutexVector = newRowsVector(frag) } else if v.fieldType == FieldTypeBool { @@ -437,12 +437,11 @@ func upgradeViewBSIv2(v *view, bitDepth uint) (ok bool, _ error) { } ok = true // mark as upgraded, requires reload - oldPath := frag.path - if newPath, err := upgradeRoaringBSIv2(frag, bitDepth); err != nil { + if tmpPath, err := upgradeRoaringBSIv2(frag, bitDepth); err != nil { return ok, errors.Wrap(err, "upgrading bsi v2") } else if err := frag.closeStorage(); err != nil { return ok, errors.Wrap(err, "closing after bsi v2 upgrade") - } else if err := os.Rename(oldPath, newPath); err != nil { + } else if err := os.Rename(tmpPath, frag.path); err != nil { return ok, errors.Wrap(err, "renaming after bsi v2 upgrade") } else if err := frag.openStorage(); err != nil { return ok, errors.Wrap(err, "re-opening after bsi v2 upgrade")