scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1684 from benbjohnson/tracing

Browse source 
Add distributed tracing.
This commit is contained in:
Ben Johnson 2018-11-21 14:53:20 -07:00 • committed by GitHub
commit 3673636902
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 765 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

205
Gopkg.lock generated
View file

@ -3,141 +3,192 @@
[[projects]]
branch = "master"
digest = "1:7deffcad61694d0cf57178c7d0c41df6b1357612e215914e35cf8b6353abf65c"
name = "github.com/CAFxX/gcnotifier"
packages = ["."]
pruneopts = ""
revision = "39b0596a2da3c92787b3319c6b5425a474b4e0da"
[[projects]]
branch = "master"
digest = "1:7519fc1e7fa9cde38634cfaf8a2ba842ea20aab2c14d81d6122ab8d6b1658d1c"
name = "github.com/DataDog/datadog-go"
packages = ["statsd"]
pruneopts = ""
revision = "ef3a9daf849df2d7ee3bbf13808dfb481069a773"
[[projects]]
digest = "1:f82b8ac36058904227087141017bb82f4b0fc58272990a4cdae3e2d6d222644e"
name = "github.com/StackExchange/wmi"
packages = ["."]
pruneopts = ""
revision = "5d049714c4a64225c3c79a7cf7d02f7fb5b96338"
version = "1.0.0"
[[projects]]
branch = "master"
digest = "1:354e62d5acb9af138e13ec842f78a846d214a8d4a9f80e578698f1f1565e2ef8"
name = "github.com/armon/go-metrics"
packages = ["."]
pruneopts = ""
revision = "3c58d8115a78a6879e5df75ae900846768d36895"
[[projects]]
digest = "1:ed112122ed4a920d944cc99b9d00b0441c11685939c28462c719488d36fe29aa"
name = "github.com/boltdb/bolt"
packages = ["."]
pruneopts = ""
revision = "2f1ce7a837dcb8da3ec595b1dac9d0632f0f99e8"
version = "v1.3.1"
[[projects]]
digest = "1:1660bb2e30cca08494f29b5593e387c6090fbe8936970ba947185b0ca000aec0"
name = "github.com/cespare/xxhash"
packages = ["."]
pruneopts = ""
revision = "5c37fe3735342a2e0d01c87a907579987c8936cc"
version = "v1.0.0"
[[projects]]
branch = "master"
digest = "1:c46fd324e7902268373e1b337436a6377c196e2dbd7b35624c6256d29d494e78"
name = "github.com/codahale/hdrhistogram"
packages = ["."]
pruneopts = ""
revision = "3a0bb77429bd3a61596f5e8a3172445844342120"
[[projects]]
digest = "1:56c130d885a4aacae1dd9c7b71cfe39912c7ebc1ff7d2b46083c8812996dc43b"
name = "github.com/davecgh/go-spew"
packages = ["spew"]
pruneopts = ""
revision = "346938d642f2ec3594ed81d874461961cd0faa76"
version = "v1.1.0"
[[projects]]
digest = "1:eb53021a8aa3f599d29c7102e65026242bdedce998a54837dc67f14b6a97c5fd"
name = "github.com/fsnotify/fsnotify"
packages = ["."]
pruneopts = ""
revision = "c2828203cd70a50dcccfb2761f8b1f8ceef9a8e9"
version = "v1.4.7"
[[projects]]
digest = "1:96c4a6ff4206086347bfe28e96e092642882128f45ecb8dc8f15f3e6f6703af0"
name = "github.com/go-ole/go-ole"
packages = [
".",
"oleutil"
"oleutil",
]
pruneopts = ""
revision = "a41e3c4b706f6ae8dfbff342b06e40fa4d2d0506"
version = "v1.2.1"
[[projects]]
digest = "1:6e73003ecd35f4487a5e88270d3ca0a81bc80dc88053ac7e4dcfec5fba30d918"
name = "github.com/gogo/protobuf"
packages = ["proto"]
pruneopts = ""
revision = "636bf0302bc95575d69441b25a2603156ffdddf1"
version = "v1.1.1"
[[projects]]
digest = "1:f958a1c137db276e52f0b50efee41a1a389dcdded59a69711f3e872757dab34b"
name = "github.com/golang/protobuf"
packages = ["proto"]
pruneopts = ""
revision = "b4deda0973fb4c70b50d226b1af49f3da59f5265"
version = "v1.1.0"
[[projects]]
digest = "1:f9f45f75f332e03fc7e9fe9188ea4e1ce4d14779ef34fa1b023da67518e36327"
name = "github.com/google/go-cmp"
packages = [
"cmp",
"cmp/cmpopts",
"cmp/internal/diff",
"cmp/internal/function",
"cmp/internal/value"
"cmp/internal/value",
]
pruneopts = ""
revision = "3af367b6b30c263d47e8895973edcca9a49cf029"
version = "v0.2.0"
[[projects]]
digest = "1:dbbeb8ddb0be949954c8157ee8439c2adfd8dc1c9510eb44a6e58cb68c3dce28"
name = "github.com/gorilla/context"
packages = ["."]
pruneopts = ""
revision = "08b5f424b9271eedf6f9f0ce86cb9396ed337a42"
version = "v1.1.1"
[[projects]]
digest = "1:a1a1522a8c1fad5675ce8ec9de96f30b898df7c8c229928c35dbd8cf277e6d62"
name = "github.com/gorilla/handlers"
packages = ["."]
pruneopts = ""
revision = "90663712d74cb411cbef281bc1e08c19d1a76145"
version = "v1.3.0"
[[projects]]
digest = "1:c2c8666b4836c81a1d247bdf21c6a6fc1ab586538ab56f74437c2e0df5c375e1"
name = "github.com/gorilla/mux"
packages = ["."]
pruneopts = ""
revision = "e3702bed27f0d39777b0b37b664b6280e8ef8fbf"
version = "v1.6.2"
[[projects]]
branch = "master"
digest = "1:4fe55793760295fbef367890352b720784243e0ad19b5ee242519a4682bb9ef8"
name = "github.com/hashicorp/errwrap"
packages = ["."]
pruneopts = ""
revision = "d6c0cd88035724dd42e0f335ae30161c20575ecc"
[[projects]]
branch = "master"
digest = "1:4423ee95d6ee30bb22f680445c58889bb5b91e1b955405bf34374a053784a8a2"
name = "github.com/hashicorp/go-immutable-radix"
packages = ["."]
pruneopts = ""
revision = "7f3cd4390caab3250a57f30efdb2a65dd7649ecf"
[[projects]]
branch = "master"
digest = "1:6396690228a7560bf9247cb90e5ae9c797bd630b01e7d2acab430bbca9a1ecb3"
name = "github.com/hashicorp/go-msgpack"
packages = ["codec"]
pruneopts = ""
revision = "fa3f63826f7c23912c15263591e65d54d080b458"
[[projects]]
branch = "master"
digest = "1:0b5ca7d18e4ded1e4dacbb37ff027cb40a80c0fed969e4e03cf7aff129bc1b44"
name = "github.com/hashicorp/go-multierror"
packages = ["."]
pruneopts = ""
revision = "3d5d8f294aa03d8e98859feac328afbdf1ae0703"
[[projects]]
branch = "master"
digest = "1:fd8ec2359315965bb6b84fd8e45cd5e8b58b80d8430dc96c8c5dfce46d30dbfc"
name = "github.com/hashicorp/go-sockaddr"
packages = ["."]
pruneopts = ""
revision = "6d291a969b86c4b633730bfc6b8b9d64c3aafed9"
[[projects]]
branch = "master"
digest = "1:9c776d7d9c54b7ed89f119e449983c3f24c0023e75001d6092442412ebca6b94"
name = "github.com/hashicorp/golang-lru"
packages = ["simplelru"]
pruneopts = ""
revision = "0fb14efe8c47ae851c0034ed7a448854d3d34cf3"
[[projects]]
branch = "master"
digest = "1:9b7c5846d70f425d7fe279595e32a20994c6075e87be03b5c367ed07280877c5"
name = "github.com/hashicorp/hcl"
packages = [
".",
@ -149,142 +200,207 @@
"hcl/token",
"json/parser",
"json/scanner",
"json/token"
"json/token",
]
pruneopts = ""
revision = "ef8a98b0bbce4a65b5aa4c368430a80ddc533168"
[[projects]]
digest = "1:d2c45a353b65012162c7ca22c39b1b0bd06d39362fb375cf42b4e48e1104bfc6"
name = "github.com/hashicorp/memberlist"
packages = ["."]
pruneopts = ""
revision = "ce8abaa0c60c2d6bee7219f5ddf500e0a1457b28"
version = "v0.1.0"
[[projects]]
digest = "1:870d441fe217b8e689d7949fef6e43efbc787e50f200cb1e70dbca9204a1d6be"
name = "github.com/inconshreveable/mousetrap"
packages = ["."]
pruneopts = ""
revision = "76626ae9c91c4f2a10f34cad8ce83ea42c93bb75"
version = "v1.0"
[[projects]]
digest = "1:961dc3b1d11f969370533390fdf203813162980c858e1dabe827b60940c909a5"
name = "github.com/magiconair/properties"
packages = ["."]
pruneopts = ""
revision = "c2353362d570a7bfa228149c62842019201cfb71"
version = "v1.8.0"
[[projects]]
digest = "1:4c8d8358c45ba11ab7bb15df749d4df8664ff1582daead28bae58cf8cbe49890"
name = "github.com/miekg/dns"
packages = ["."]
pruneopts = ""
revision = "5a2b9fab83ff0f8bfc99684bd5f43a37abe560f1"
version = "v1.0.8"
[[projects]]
branch = "master"
digest = "1:f43ed2c836208c14f45158fd01577c985688a4d11cf9fd475a939819fef3b321"
name = "github.com/mitchellh/mapstructure"
packages = ["."]
pruneopts = ""
revision = "f15292f7a699fcc1a38a80977f80a046874ba8ac"
[[projects]]
digest = "1:78fb99d6011c2ae6c72f3293a83951311147b12b06a5ffa43abf750c4fab6ac5"
name = "github.com/opentracing/opentracing-go"
packages = [
".",
"ext",
"log",
]
pruneopts = ""
revision = "1949ddbfd147afd4d964a9f00b24eb291e0e7c38"
version = "v1.0.2"
[[projects]]
digest = "1:894aef961c056b6d85d12bac890bf60c44e99b46292888bfa66caf529f804457"
name = "github.com/pelletier/go-toml"
packages = ["."]
pruneopts = ""
revision = "c01d1270ff3e442a8a57cddc1c92dc1138598194"
version = "v1.2.0"
[[projects]]
name = "github.com/pilosa/go-pilosa"
packages = [
".",
"gopilosa_pbuf"
]
revision = "4e7807f5ad779407936744057cd17332046b6c3c"
version = "v1.1.0"
[[projects]]
digest = "1:7365acd48986e205ccb8652cc746f09c8b7876030d53710ea6ef7d0bd0dcd7ca"
name = "github.com/pkg/errors"
packages = ["."]
pruneopts = ""
revision = "645ef00459ed84a119197bfb8d8205042c6df63d"
version = "v0.8.0"
[[projects]]
digest = "1:7f569d906bdd20d906b606415b7d794f798f91a62fcfb6a4daa6d50690fb7a3f"
name = "github.com/satori/go.uuid"
packages = ["."]
pruneopts = ""
revision = "f58768cc1a7a7e77a3bd49e98cdd21419399b6a3"
version = "v1.2.0"
[[projects]]
branch = "master"
digest = "1:6ee36f2cea425916d81fdaaf983469fc18f91b3cf090cfe90fa0a9d85b8bfab7"
name = "github.com/sean-/seed"
packages = ["."]
pruneopts = ""
revision = "e2103e2c35297fb7e17febb81e49b312087a2372"
[[projects]]
digest = "1:02715a2fb4b9279af36651a59a51dd4164eb689bd6785874811899f43eeb2a54"
name = "github.com/shirou/gopsutil"
packages = [
"cpu",
"host",
"internal/common",
"mem",
"net",
"process"
"process",
]
pruneopts = ""
revision = "8048a2e9c5773235122027dd585cf821b2af1249"
version = "v2.18.07"
[[projects]]
branch = "master"
name = "github.com/shirou/w32"
packages = ["."]
revision = "bb4de0191aa41b5507caa14b0650cdbddcd9280b"
[[projects]]
digest = "1:7ba2551c9a8de293bc575dbe2c0d862c52252d26f267f784547f059f512471c8"
name = "github.com/spf13/afero"
packages = [
".",
"mem"
"mem",
]
pruneopts = ""
revision = "787d034dfe70e44075ccc060d346146ef53270ad"
version = "v1.1.1"
[[projects]]
digest = "1:d0b38ba6da419a6d4380700218eeec8623841d44a856bb57369c172fbf692ab4"
name = "github.com/spf13/cast"
packages = ["."]
pruneopts = ""
revision = "8965335b8c7107321228e3e3702cab9832751bac"
version = "v1.2.0"
[[projects]]
digest = "1:a1403cc8a94b8d7956ee5e9694badef0e7b051af289caad1cf668331e3ffa4f6"
name = "github.com/spf13/cobra"
packages = ["."]
pruneopts = ""
revision = "ef82de70bb3f60c65fb8eebacbb2d122ef517385"
version = "v0.0.3"
[[projects]]
branch = "master"
digest = "1:104517520aab91164020ab6524a5d6b7cafc641b2e42ac6236f6ac1deac4f66a"
name = "github.com/spf13/jwalterweatherman"
packages = ["."]
pruneopts = ""
revision = "7c0cea34c8ece3fbeb2b27ab9b59511d360fb394"
[[projects]]
digest = "1:8e243c568f36b09031ec18dff5f7d2769dcf5ca4d624ea511c8e3197dc3d352d"
name = "github.com/spf13/pflag"
packages = ["."]
pruneopts = ""
revision = "583c0c0531f06d5278b7d917446061adc344b5cd"
version = "v1.0.1"
[[projects]]
digest = "1:3dab237cd3263a290d771d133fed777bb56c22e380b00ebe92e6531d5c8d3d0c"
name = "github.com/spf13/viper"
packages = ["."]
pruneopts = ""
revision = "b5e8006cbee93ec955a89ab31e0e3ce3204f3736"
version = "v1.0.2"
[[projects]]
digest = "1:941ab4973b3218a9a6d02d31734f5c762239eb538c0d702fff62d4037af8ab0a"
name = "github.com/uber/jaeger-client-go"
packages = [
".",
"config",
"internal/baggage",
"internal/baggage/remote",
"internal/spanlog",
"internal/throttler",
"internal/throttler/remote",
"log",
"rpcmetrics",
"thrift",
"thrift-gen/agent",
"thrift-gen/baggage",
"thrift-gen/jaeger",
"thrift-gen/sampling",
"thrift-gen/zipkincore",
"transport",
"utils",
]
pruneopts = ""
revision = "1a782e2da844727691fef1757c72eb190c2909f0"
version = "v2.15.0"
[[projects]]
digest = "1:aa1598d34009b45ce74fdabdd25e4258d7923d1e1b418d4c98482e79607cb9b0"
name = "github.com/uber/jaeger-lib"
packages = ["metrics"]
pruneopts = ""
revision = "ed3a127ec5fef7ae9ea95b01b542c47fbd999ce5"
version = "v1.5.0"
[[projects]]
branch = "master"
digest = "1:cae234a803b78380e4d769db6036b9fcc8c08ed4ff862571ffc1a958edc1f629"
name = "golang.org/x/crypto"
packages = [
"ed25519",
"ed25519/internal/edwards25519"
"ed25519/internal/edwards25519",
]
pruneopts = ""
revision = "c126467f60eb25f8f27e5a981f32a87e3965053f"
[[projects]]
branch = "master"
digest = "1:1f3b488bf9c50cac0cb738fca05d8ff55509c8698f58ea4d74b1499d0baaeb74"
name = "golang.org/x/net"
packages = [
"bpf",
@ -292,26 +408,32 @@
"internal/iana",
"internal/socket",
"ipv4",
"ipv6"
"ipv6",
]
pruneopts = ""
revision = "22bb95c5e783d192c577a7b310b06637db9f1d94"
[[projects]]
branch = "master"
digest = "1:b2ea75de0ccb2db2ac79356407f8a4cd8f798fe15d41b381c00abf3ae8e55ed1"
name = "golang.org/x/sync"
packages = ["errgroup"]
pruneopts = ""
revision = "1d60e4601c6fd243af51cc01ddf169918a5407ca"
[[projects]]
branch = "master"
digest = "1:dd631ee90bd2e7aa16b6e094217d77a797684b52811374c948c695cbb46b5bbb"
name = "golang.org/x/sys"
packages = [
"unix",
"windows"
"windows",
]
pruneopts = ""
revision = "bd9dbc187b6e1dacfdd2722a87e83093c2d7bd6e"
[[projects]]
digest = "1:5acd3512b047305d49e8763eef7ba423901e85d5dd2fd1e71778a0ea8de10bd4"
name = "golang.org/x/text"
packages = [
"internal/gen",
@ -319,20 +441,49 @@
"internal/ucd",
"transform",
"unicode/cldr",
"unicode/norm"
"unicode/norm",
]
pruneopts = ""
revision = "f21a4dfb5e38f5895301dc265a8def02365cc3d0"
version = "v0.3.0"
[[projects]]
digest = "1:f0620375dd1f6251d9973b5f2596228cc8042e887cd7f827e4220bc1ce8c30e2"
name = "gopkg.in/yaml.v2"
packages = ["."]
pruneopts = ""
revision = "5420a8b6744d3b0345ab293f6fcba19c978f1183"
version = "v2.2.1"
[solve-meta]
analyzer-name = "dep"
analyzer-version = 1
inputs-digest = "be318fa4f2a72e7e849b2faff1ce9300deaaf2d24cf76100f4b538abed86295b"
input-imports = [
"github.com/CAFxX/gcnotifier",
"github.com/DataDog/datadog-go/statsd",
"github.com/boltdb/bolt",
"github.com/cespare/xxhash",
"github.com/davecgh/go-spew/spew",
"github.com/gogo/protobuf/proto",
"github.com/golang/protobuf/proto",
"github.com/google/go-cmp/cmp",
"github.com/google/go-cmp/cmp/cmpopts",
"github.com/gorilla/handlers",
"github.com/gorilla/mux",
"github.com/hashicorp/memberlist",
"github.com/opentracing/opentracing-go",
"github.com/opentracing/opentracing-go/ext",
"github.com/pelletier/go-toml",
"github.com/pkg/errors",
"github.com/satori/go.uuid",
"github.com/shirou/gopsutil/host",
"github.com/shirou/gopsutil/mem",
"github.com/spf13/cobra",
"github.com/spf13/pflag",
"github.com/spf13/viper",
"github.com/uber/jaeger-client-go",
"github.com/uber/jaeger-client-go/config",
"golang.org/x/sync/errgroup",
]
solver-name = "gps-cdcl"
solver-version = 1

122
api.go
View file

@ -29,6 +29,7 @@ import (
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/roaring"
"github.com/pilosa/pilosa/stats"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -99,6 +100,9 @@ func (api *API) validate(f apiMethod) error {
// Query parses a PQL query out of the request and executes it.
func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "API.Query")
defer span.Finish()
if err := api.validate(apiQuery); err != nil {
return QueryResponse{}, errors.Wrap(err, "validating api method")
}
@ -122,7 +126,10 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
// CreateIndex makes a new Pilosa index.
func (api *API) CreateIndex(_ context.Context, indexName string, options IndexOptions) (*Index, error) {
func (api *API) CreateIndex(ctx context.Context, indexName string, options IndexOptions) (*Index, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.CreateIndex")
defer span.Finish()
if err := api.validate(apiCreateIndex); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -146,7 +153,10 @@ func (api *API) CreateIndex(_ context.Context, indexName string, options IndexOp
}
// Index retrieves the named index.
func (api *API) Index(_ context.Context, indexName string) (*Index, error) {
func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Index")
defer span.Finish()
if err := api.validate(apiIndex); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -160,7 +170,10 @@ func (api *API) Index(_ context.Context, indexName string) (*Index, error) {
// DeleteIndex removes the named index. If the index is not found it does
// nothing and returns no error.
func (api *API) DeleteIndex(_ context.Context, indexName string) error {
func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.DeleteIndex")
defer span.Finish()
if err := api.validate(apiDeleteIndex); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -186,7 +199,10 @@ func (api *API) DeleteIndex(_ context.Context, indexName string) error {
// CreateField makes the named field in the named index with the given options.
// This method currently only takes a single functional option, but that may be
// changed in the future to support multiple options.
func (api *API) CreateField(_ context.Context, indexName string, fieldName string, opts ...FieldOption) (*Field, error) {
func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, opts ...FieldOption) (*Field, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.CreateField")
defer span.Finish()
if err := api.validate(apiCreateField); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -228,7 +244,10 @@ func (api *API) CreateField(_ context.Context, indexName string, fieldName strin
}
// Field retrieves the named field.
func (api *API) Field(_ context.Context, indexName, fieldName string) (*Field, error) {
func (api *API) Field(ctx context.Context, indexName, fieldName string) (*Field, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Field")
defer span.Finish()
if err := api.validate(apiField); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -269,6 +288,9 @@ func setUpImportOptions(opts ...ImportOption) (*ImportOptions, error) {
// of the rows in this shard of this field concatenated together in one long
// bitmap.
func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, shard uint64, remote bool, req *ImportRoaringRequest) (err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "API.ImportRoaring")
defer span.Finish()
if err = api.validate(apiField); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -326,7 +348,10 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string,
// DeleteField removes the named field from the named index. If the index is not
// found, an error is returned. If the field is not found, it is ignored and no
// action is taken.
func (api *API) DeleteField(_ context.Context, indexName string, fieldName string) error {
func (api *API) DeleteField(ctx context.Context, indexName string, fieldName string) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.DeleteField")
defer span.Finish()
if err := api.validate(apiDeleteField); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -390,7 +415,10 @@ func (api *API) DeleteAvailableShard(_ context.Context, indexName, fieldName str
// ExportCSV encodes the fragment designated by the index,field,shard as
// CSV of the form <row>,<col>
func (api *API) ExportCSV(_ context.Context, indexName string, fieldName string, shard uint64, w io.Writer) error {
func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName string, shard uint64, w io.Writer) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.ExportCSV")
defer span.Finish()
if err := api.validate(apiExportCSV); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -424,6 +452,7 @@ func (api *API) ExportCSV(_ context.Context, indexName string, fieldName string,
// Define the function to write each bit as a string,
// translating to keys where necessary.
var n int
fn := func(rowID, columnID uint64) error {
var rowStr string
var colStr string
@ -445,6 +474,7 @@ func (api *API) ExportCSV(_ context.Context, indexName string, fieldName string,
colStr = strconv.FormatUint(columnID, 10)
}
n++
return cw.Write([]string{rowStr, colStr})
}
@ -456,11 +486,16 @@ func (api *API) ExportCSV(_ context.Context, indexName string, fieldName string,
// Ensure data is flushed.
cw.Flush()
span.LogKV("n", n)
return nil
}
// ShardNodes returns the node and all replicas which should contain a shard's data.
func (api *API) ShardNodes(_ context.Context, indexName string, shard uint64) ([]*Node, error) {
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*Node, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.ShardNodes")
defer span.Finish()
if err := api.validate(apiShardNodes); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -471,7 +506,10 @@ func (api *API) ShardNodes(_ context.Context, indexName string, shard uint64) ([
// FragmentBlockData is an endpoint for internal usage. It is not guaranteed to
// return anything useful. Currently it returns protobuf encoded row and column
// ids from a "block" which is a subdivision of a fragment.
func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, error) {
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlockData")
defer span.Finish()
if err := api.validate(apiFragmentBlockData); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -504,7 +542,10 @@ func (api *API) FragmentBlockData(_ context.Context, body io.Reader) ([]byte, er
}
// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment.
func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) {
func (api *API) FragmentBlocks(ctx context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlocks")
defer span.Finish()
if err := api.validate(apiFragmentBlocks); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -522,7 +563,9 @@ func (api *API) FragmentBlocks(_ context.Context, indexName, fieldName, viewName
// Hosts returns a list of the hosts in the cluster including their ID,
// URL, and which is the coordinator.
func (api *API) Hosts(_ context.Context) []*Node {
func (api *API) Hosts(ctx context.Context) []*Node {
span, _ := tracing.StartSpanFromContext(ctx, "API.Hosts")
defer span.Finish()
return api.cluster.Nodes()
}
@ -533,7 +576,10 @@ func (api *API) Node() *Node {
}
// RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests.
func (api *API) RecalculateCaches(_ context.Context) error {
func (api *API) RecalculateCaches(ctx context.Context) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.RecalculateCaches")
defer span.Finish()
if err := api.validate(apiRecalculateCaches); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -548,7 +594,10 @@ func (api *API) RecalculateCaches(_ context.Context) error {
// PostClusterMessage is for internal use. It decodes a protobuf message out of
// the body and forwards it to the BroadcastHandler.
func (api *API) ClusterMessage(_ context.Context, reqBody io.Reader) error {
func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.ClusterMessage")
defer span.Finish()
if err := api.validate(apiClusterMessage); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -575,12 +624,17 @@ func (api *API) ClusterMessage(_ context.Context, reqBody io.Reader) error {
// Schema returns information about each index in Pilosa including which fields
// they contain.
func (api *API) Schema(_ context.Context) []*IndexInfo {
func (api *API) Schema(ctx context.Context) []*IndexInfo {
span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
defer span.Finish()
return api.holder.limitedSchema()
}
// Views returns the views in the given field.
func (api *API) Views(_ context.Context, indexName string, fieldName string) ([]*view, error) {
func (api *API) Views(ctx context.Context, indexName string, fieldName string) ([]*view, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Views")
defer span.Finish()
if err := api.validate(apiViews); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -597,7 +651,10 @@ func (api *API) Views(_ context.Context, indexName string, fieldName string) ([]
}
// DeleteView removes the given view.
func (api *API) DeleteView(_ context.Context, indexName string, fieldName string, viewName string) error {
func (api *API) DeleteView(ctx context.Context, indexName string, fieldName string, viewName string) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.DeleteView")
defer span.Finish()
if err := api.validate(apiDeleteView); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -631,7 +688,10 @@ func (api *API) DeleteView(_ context.Context, indexName string, fieldName string
}
// IndexAttrDiff
func (api *API) IndexAttrDiff(_ context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.IndexAttrDiff")
defer span.Finish()
if err := api.validate(apiIndexAttrDiff); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -665,7 +725,10 @@ func (api *API) IndexAttrDiff(_ context.Context, indexName string, blocks []Attr
return attrs, nil
}
func (api *API) FieldAttrDiff(_ context.Context, indexName string, fieldName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (api *API) FieldAttrDiff(ctx context.Context, indexName string, fieldName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FieldAttrDiff")
defer span.Finish()
if err := api.validate(apiFieldAttrDiff); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -724,6 +787,9 @@ func OptImportOptionsIgnoreKeyCheck(b bool) ImportOption {
// Import bulk imports data into a particular index,field,shard.
func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOption) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.Import")
defer span.Finish()
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -829,6 +895,9 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts ...ImportOption) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.ImportValue")
defer span.Finish()
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -922,7 +991,10 @@ func importExistenceColumns(index *Index, columnIDs []uint64) error {
// MaxShards returns the maximum shard number for each index in a map.
// TODO (2.0): This method has been deprecated. Instead, use
// AvailableShardsByIndex.
func (api *API) MaxShards(_ context.Context) map[string]uint64 {
func (api *API) MaxShards(ctx context.Context) map[string]uint64 {
span, _ := tracing.StartSpanFromContext(ctx, "API.MaxShards")
defer span.Finish()
m := make(map[string]uint64)
for k, v := range api.holder.availableShardsByIndex() {
m[k] = v.Max()
@ -931,7 +1003,9 @@ func (api *API) MaxShards(_ context.Context) map[string]uint64 {
}
// AvailableShardsByIndex returns bitmaps of shards with available by index name.
func (api *API) AvailableShardsByIndex(_ context.Context) map[string]*roaring.Bitmap {
func (api *API) AvailableShardsByIndex(ctx context.Context) map[string]*roaring.Bitmap {
span, _ := tracing.StartSpanFromContext(ctx, "API.AvailableShardsByIndex")
defer span.Finish()
return api.holder.availableShardsByIndex()
}
@ -982,7 +1056,10 @@ func (api *API) indexField(indexName string, fieldName string, shard uint64) (*I
}
// SetCoordinator makes a new Node the cluster coordinator.
func (api *API) SetCoordinator(_ context.Context, id string) (oldNode, newNode *Node, err error) {
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *Node, err error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.SetCoordinator")
defer span.Finish()
if err := api.validate(apiSetCoordinator); err != nil {
return nil, nil, errors.Wrap(err, "validating api method")
}
@ -1047,6 +1124,9 @@ func (api *API) ResizeAbort() error {
// GetTranslateData provides a reader for key translation logs starting at offset.
func (api *API) GetTranslateData(ctx context.Context, offset int64) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "API.GetTranslateData")
defer span.Finish()
rc, err := api.holder.translateFile.Reader(ctx, offset)
if err != nil {
return nil, errors.Wrap(err, "read from translate store")

22
attr.go
View file

@ -203,25 +203,3 @@ func DecodeAttrs(v []byte) (map[string]interface{}, error) {
}
return decodeAttrs(pb.GetAttrs()), nil
}
// memAttrStore represents an in-memory implementation of the AttrStore interface.
type memAttrStore struct {
store map[uint64]map[string]interface{}
}
func (s *memAttrStore) Path() string { return "" }
func (s *memAttrStore) Open() error { return nil }
func (s *memAttrStore) Close() error { return nil }
func (s *memAttrStore) Attrs(id uint64) (m map[string]interface{}, err error) { return s.store[id], nil }
func (s *memAttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
s.store[id] = m
return nil
}
func (s *memAttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
for id, v := range m {
s.store[id] = v
}
return nil
}
func (s *memAttrStore) Blocks() ([]AttrBlock, error) { return nil, nil }
func (s *memAttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) { return nil, nil }

8
cluster.go
View file

@ -27,14 +27,14 @@ import (
"sync"
"time"
"golang.org/x/sync/errgroup"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/logger"
"github.com/pilosa/pilosa/roaring"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
"golang.org/x/sync/errgroup"
)
const (
@ -1260,6 +1260,8 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
// Stop processing on any error.
if err := func() error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "Cluster.followResizeInstruction")
defer span.Finish()
// Sync the schema received in the resize instruction.
c.logger.Debugf("holder applySchema")
@ -1293,7 +1295,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
// Stream shard from remote node.
c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
rd, err := c.InternalClient.RetrieveShardFromURI(context.Background(), src.Index, src.Field, src.Shard, srcURI)
rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.Shard, srcURI)
if err != nil {
// For now it is an acceptable error if the fragment is not found
// on the remote node. This occurs when a shard has been skipped and

27
cmd/server.go
View file

@ -17,11 +17,13 @@ package cmd
import (
"io"
"github.com/pkg/errors"
"github.com/spf13/cobra"
"github.com/pilosa/pilosa/ctl"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/tracing"
"github.com/pilosa/pilosa/tracing/opentracing"
"github.com/pkg/errors"
"github.com/spf13/cobra"
jaegercfg "github.com/uber/jaeger-client-go/config"
)
// Server is global so that tests can control and verify it.
@ -39,9 +41,28 @@ It will load existing data from the configured
directory and start listening for client connections
on the configured port.`,
RunE: func(cmd *cobra.Command, args []string) error {
// Start & run the server.
if err := Server.Start(); err != nil {
return errors.Wrap(err, "running server")
}
// Initialize tracing in the command since it is global.
var cfg jaegercfg.Configuration
cfg.ServiceName = "pilosa"
cfg.Sampler = &jaegercfg.SamplerConfig{
Type: Server.Config.Tracing.SamplerType,
Param: Server.Config.Tracing.SamplerParam,
}
cfg.Reporter = &jaegercfg.ReporterConfig{
LocalAgentHostPort: Server.Config.Tracing.AgentHostPort,
}
tracer, closer, err := cfg.NewTracer()
if err != nil {
return errors.Wrap(err, "initializing jaeger tracer")
}
defer closer.Close()
tracing.GlobalTracer = opentracing.NewTracer(tracer)
return errors.Wrap(Server.Wait(), "waiting on Server")
},
}

5
ctl/server.go
View file

@ -68,4 +68,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", srv.Config.Metric.Host, "Default URI to send metrics.")
flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", (time.Duration)(srv.Config.Metric.PollInterval), "Polling interval metrics.")
flags.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", srv.Config.Metric.Diagnostics, "Enabled diagnostics reporting.")
// Tracing
flags.StringVarP(&srv.Config.Tracing.AgentHostPort, "tracing.agent-host-port", "", srv.Config.Tracing.AgentHostPort, "Jaeger agent host:port.")
flags.StringVarP(&srv.Config.Tracing.SamplerType, "tracing.sampler-type", "", srv.Config.Tracing.SamplerType, "Jaeger sampler type.")
flags.Float64VarP(&srv.Config.Tracing.SamplerParam, "tracing.sampler-param", "", srv.Config.Tracing.SamplerParam, "Jaeger sampler parameter.")
}

36
docs/configuration.md
View file

@ -307,6 +307,42 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
skip-verify = true
```
#### Tracing Sampler Type
* Description: Jaeger sampler type (const, probabilistic, ratelimiting, or remote)
* Flag: `tracing.sampler-type`
* Env: `PILOSA_TRACING_SAMPLER_TYPE`
* Config:
```toml
[tracing]
sampler-type = "remote"
```
#### Tracing Sampler Parameter
* Description: Jaeger sampler parameter (number)
* Flag: `tracing.sampler-param`
* Env: `PILOSA_TRACING_SAMPLER_PARAM`
* Config:
```toml
[tracing]
sampler-param = 0.001
```
#### Tracing Agent Host/Port
* Description: Jaeger agent host:port
* Flag: `tracing.agent-host-port`
* Env: `PILOSA_TRACING_AGENT_HOST_PORT`
* Config:
```toml
[tracing]
agent-host-port = "localhost:6831"
```
#### Translation Map Size
* Description: Size in bytes of mmap to allocate for key translation

162
executor.go
View file

@ -22,6 +22,7 @@ import (
"time"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
)
@ -81,7 +82,11 @@ func newExecutor(opts ...executorOption) *executor {
// Execute executes a PQL query.
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) (QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.Execute")
defer span.Finish()
resp := QueryResponse{}
// Verify that an index is set.
if index == "" {
return resp, ErrIndexRequired
@ -105,10 +110,8 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// Translate query keys to ids, if necessary.
// No need to translate a remote call.
if !opt.Remote {
for i := range q.Calls {
if err := e.translateCall(index, idx, q.Calls[i]); err != nil {
return resp, err
}
if err := e.translateCalls(ctx, index, idx, q.Calls); err != nil {
return resp, err
}
}
@ -154,11 +157,8 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// Translate response objects from ids to keys, if necessary.
// No need to translate a remote call.
if !opt.Remote {
for i := range results {
results[i], err = e.translateResult(index, idx, q.Calls[i], results[i])
if err != nil {
return resp, err
}
if err := e.translateResults(ctx, index, idx, q.Calls, results); err != nil {
return resp, err
}
}
@ -189,6 +189,9 @@ func (e *executor) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttr
}
func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) ([]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.execute")
defer span.Finish()
// Don't bother calculating shards for query types that don't require it.
needsShards := needsShards(q.Calls)
@ -225,6 +228,9 @@ func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shar
// executeCall executes a call.
func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeCall")
defer span.Finish()
if err := e.validateCallArgs(c); err != nil {
return nil, errors.Wrap(err, "validating args")
}
@ -292,6 +298,9 @@ func (e *executor) validateCallArgs(c *pql.Call) error {
}
func (e *executor) executeOptionsCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeOptionsCall")
defer span.Finish()
optCopy := &execOptions{}
*optCopy = *opt
if arg, ok := c.Args["columnAttrs"]; ok {
@ -335,6 +344,9 @@ func (e *executor) executeOptionsCall(ctx context.Context, index string, c *pql.
// executeSum executes a Sum() call.
func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSum")
defer span.Finish()
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Sum(): field required")
}
@ -368,6 +380,9 @@ func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, sh
// executeMin executes a Min() call.
func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMin")
defer span.Finish()
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Min(): field required")
}
@ -401,6 +416,9 @@ func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, sh
// executeMax executes a Max() call.
func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMax")
defer span.Finish()
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Max(): field required")
}
@ -434,6 +452,9 @@ func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, sh
// executeBitmapCall executes a call that returns a bitmap.
func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCall")
defer span.Finish()
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeBitmapCallShard(ctx, index, c, shard)
@ -500,6 +521,9 @@ func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
// executeBitmapCallShard executes a bitmap call for a single shard.
func (e *executor) executeBitmapCallShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCallShard")
defer span.Finish()
switch c.Name {
case "Row":
return e.executeBitmapShard(ctx, index, c, shard)
@ -522,6 +546,9 @@ func (e *executor) executeBitmapCallShard(ctx context.Context, index string, c *
// executeSumCountShard calculates the sum and count for bsiGroups on a shard.
func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSumCountShard")
defer span.Finish()
var filter *Row
if len(c.Children) == 1 {
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
@ -560,6 +587,9 @@ func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pq
// executeMinShard calculates the min for bsiGroups on a shard.
func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinShard")
defer span.Finish()
var filter *Row
if len(c.Children) == 1 {
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
@ -598,6 +628,9 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal
// executeMaxShard calculates the max for bsiGroups on a shard.
func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxShard")
defer span.Finish()
var filter *Row
if len(c.Children) == 1 {
row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
@ -638,6 +671,9 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
// This first performs the TopN() to determine the top results and then
// requeries to retrieve the full counts for each of the top results.
func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeTopN")
defer span.Finish()
idsArg, _, err := c.UintSliceArg("ids")
if err != nil {
return nil, fmt.Errorf("executeTopN: %v", err)
@ -677,6 +713,9 @@ func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, s
}
func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeTopNShards")
defer span.Finish()
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeTopNShard(ctx, index, c, shard)
@ -702,6 +741,9 @@ func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.C
// executeTopNShard executes a TopN call for a single shard.
func (e *executor) executeTopNShard(ctx context.Context, index string, c *pql.Call, shard uint64) ([]Pair, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeTopNShard")
defer span.Finish()
field, _ := c.Args["_field"].(string)
n, _, err := c.UintArg("n")
if err != nil {
@ -764,6 +806,9 @@ func (e *executor) executeTopNShard(ctx context.Context, index string, c *pql.Ca
// executeDifferenceShard executes a difference() call for a local shard.
func (e *executor) executeDifferenceShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeDifferenceShard")
defer span.Finish()
var other *Row
if len(c.Children) == 0 {
return nil, fmt.Errorf("empty Difference query is currently not supported")
@ -1093,8 +1138,11 @@ func (e *executor) executeRowsShard(_ context.Context, index string, c *pql.Call
return frag.rows(start, filters...), nil
}
func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
// Fetch index.
func (e *executor) executeBitmapShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapShard")
defer span.Finish()
// Fetch column label from index.
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
@ -1126,6 +1174,9 @@ func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Ca
// executeIntersectShard executes a intersect() call for a local shard.
func (e *executor) executeIntersectShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeIntersectShard")
defer span.Finish()
var other *Row
if len(c.Children) == 0 {
return nil, fmt.Errorf("empty Intersect query is currently not supported")
@ -1148,6 +1199,9 @@ func (e *executor) executeIntersectShard(ctx context.Context, index string, c *p
// executeRangeShard executes a range() call for a local shard.
func (e *executor) executeRangeShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeRangeShard")
defer span.Finish()
// Handle bsiGroup ranges differently.
if c.HasConditionArg() {
return e.executeBSIGroupRangeShard(ctx, index, c, shard)
@ -1220,7 +1274,10 @@ func (e *executor) executeRangeShard(ctx context.Context, index string, c *pql.C
}
// executeBSIGroupRangeShard executes a range(bsiGroup) call for a local shard.
func (e *executor) executeBSIGroupRangeShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeBSIGroupRangeShard")
defer span.Finish()
// Only one conditional should be present.
if len(c.Args) == 0 {
return nil, errors.New("Range(): condition required")
@ -1351,6 +1408,9 @@ func (e *executor) executeBSIGroupRangeShard(_ context.Context, index string, c
// executeUnionShard executes a union() call for a local shard.
func (e *executor) executeUnionShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeUnionShard")
defer span.Finish()
other := NewRow()
for i, input := range c.Children {
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
@ -1370,6 +1430,9 @@ func (e *executor) executeUnionShard(ctx context.Context, index string, c *pql.C
// executeXorShard executes a xor() call for a local shard.
func (e *executor) executeXorShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeXorShard")
defer span.Finish()
other := NewRow()
for i, input := range c.Children {
row, err := e.executeBitmapCallShard(ctx, index, input, shard)
@ -1389,6 +1452,9 @@ func (e *executor) executeXorShard(ctx context.Context, index string, c *pql.Cal
// executeNotShard executes a not() call for a local shard.
func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeNotShard")
defer span.Finish()
if len(c.Children) == 0 {
return nil, errors.New("Not() requires an input row")
} else if len(c.Children) > 1 {
@ -1421,6 +1487,9 @@ func (e *executor) executeNotShard(ctx context.Context, index string, c *pql.Cal
// executeCount executes a count() call.
func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeCount")
defer span.Finish()
if len(c.Children) == 0 {
return 0, errors.New("Count() requires an input bitmap")
} else if len(c.Children) > 1 {
@ -1453,6 +1522,9 @@ func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call,
// executeClearBit executes a Clear() call.
func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearBit")
defer span.Finish()
fieldName, err := c.FieldArg()
if err != nil {
return false, errors.New("Clear() argument required: field")
@ -1488,6 +1560,9 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
// executeClearBitField executes a Clear() call for a field.
func (e *executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearBitField")
defer span.Finish()
shard := colID / ShardWidth
ret := false
for _, node := range e.Cluster.shardNodes(index, shard) {
@ -1518,6 +1593,9 @@ func (e *executor) executeClearBitField(ctx context.Context, index string, c *pq
// executeClearRow executes a ClearRow() call.
func (e *executor) executeClearRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeClearRow")
defer span.Finish()
// Ensure the field type supports ClearRow().
fieldName, err := c.FieldArg()
if err != nil {
@ -1554,7 +1632,10 @@ func (e *executor) executeClearRow(ctx context.Context, index string, c *pql.Cal
}
// executeClearRowShard executes a ClearRow() call for a single shard.
func (e *executor) executeClearRowShard(_ context.Context, index string, c *pql.Call, shard uint64) (bool, error) {
func (e *executor) executeClearRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (bool, error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeClearRowShard")
defer span.Finish()
fieldName, err := c.FieldArg()
if err != nil {
return false, errors.New("ClearRow() argument required: field")
@ -1680,6 +1761,9 @@ func (e *executor) executeSetRowShard(ctx context.Context, index string, c *pql.
// executeSet executes a Set() call.
func (e *executor) executeSet(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSet")
defer span.Finish()
// Read colID.
colID, ok, err := c.UintArg("_" + columnLabel)
if err != nil {
@ -1747,6 +1831,9 @@ func (e *executor) executeSet(ctx context.Context, index string, c *pql.Call, op
// executeSetBitField executes a Set() call for a specific field.
func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetBitField")
defer span.Finish()
shard := colID / ShardWidth
ret := false
@ -1779,6 +1866,9 @@ func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.
// executeSetValueField executes a Set() call for a specific int field.
func (e *executor) executeSetValueField(ctx context.Context, index string, c *pql.Call, f *Field, colID uint64, value int64, opt *execOptions) (bool, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetValueField")
defer span.Finish()
shard := colID / ShardWidth
ret := false
@ -1811,6 +1901,9 @@ func (e *executor) executeSetValueField(ctx context.Context, index string, c *pq
// executeSetRowAttrs executes a SetRowAttrs() call.
func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetRowAttrs")
defer span.Finish()
fieldName, ok := c.Args["_field"].(string)
if !ok {
return errors.New("SetRowAttrs() field required")
@ -1868,6 +1961,9 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.
// executeBulkSetRowAttrs executes a set of SetRowAttrs() calls.
func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, calls []*pql.Call, opt *execOptions) ([]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBulkSetRowAttrs")
defer span.Finish()
// Collect attributes by field/id.
m := make(map[string]map[uint64]map[string]interface{})
for _, c := range calls {
@ -1955,6 +2051,9 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal
// executeSetColumnAttrs executes a SetColumnAttrs() call.
func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeSetColumnAttrs")
defer span.Finish()
// Retrieve index.
idx := e.Holder.Index(index)
if idx == nil {
@ -2003,6 +2102,9 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p
// remoteExec executes a PQL query remotely for a set of shards on a node.
func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *pql.Query, shards []uint64) (results []interface{}, err error) { // nolint: interfacer
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeExec")
defer span.Finish()
// Encode request object.
pbreq := &QueryRequest{
Query: q.String(),
@ -2041,6 +2143,9 @@ loop:
// If a mapping of shards to a node fails then the shards are resplit across
// secondary nodes and retried. This continues to occur until all nodes are exhausted.
func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapReduce")
defer span.Finish()
ch := make(chan mapResponse)
// Wrap context with a cancel to kill goroutines on exit.
@ -2100,6 +2205,9 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
}
func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
defer span.Finish()
// Group shards together by nodes.
m, err := e.shardsByNode(nodes, index, shards)
if err != nil {
@ -2135,6 +2243,9 @@ func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
// mapperLocal performs map & reduce entirely on the local node.
func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
defer span.Finish()
ch := make(chan mapResponse, len(shards))
for _, shard := range shards {
@ -2171,6 +2282,18 @@ func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFu
}
}
func (e *executor) translateCalls(ctx context.Context, index string, idx *Index, calls []*pql.Call) error {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.translateCalls")
defer span.Finish()
for i := range calls {
if err := e.translateCall(index, idx, calls[i]); err != nil {
return err
}
}
return nil
}
func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
var colKey, rowKey, fieldName string
switch c.Name {
@ -2319,6 +2442,19 @@ func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) e
return nil
}
func (e *executor) translateResults(ctx context.Context, index string, idx *Index, calls []*pql.Call, results []interface{}) (err error) {
span, _ := tracing.StartSpanFromContext(ctx, "Executor.translateResults")
defer span.Finish()
for i := range results {
results[i], err = e.translateResult(index, idx, calls[i], results[i])
if err != nil {
return err
}
}
return nil
}
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
switch result := result.(type) {
case *Row:

13
fragment.go
View file

@ -40,6 +40,7 @@ import (
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/roaring"
"github.com/pilosa/pilosa/stats"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
)
@ -2187,6 +2188,9 @@ func (s *fragmentSyncer) isClosing() bool {
// syncFragment compares checksums for the local and remote fragments and
// then merges any blocks which have differences.
func (s *fragmentSyncer) syncFragment() error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncFragment")
defer span.Finish()
// Determine replica set.
nodes := s.Cluster.shardNodes(s.Fragment.index, s.Fragment.shard)
if len(nodes) == 1 {
@ -2204,7 +2208,7 @@ func (s *fragmentSyncer) syncFragment() error {
}
// Retrieve remote blocks.
blocks, err := s.Cluster.InternalClient.FragmentBlocks(context.Background(), &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.view, s.Fragment.shard)
blocks, err := s.Cluster.InternalClient.FragmentBlocks(ctx, &node.URI, s.Fragment.index, s.Fragment.field, s.Fragment.view, s.Fragment.shard)
if err != nil && err != ErrFragmentNotFound {
return errors.Wrap(err, "getting blocks")
}
@ -2264,6 +2268,9 @@ func (s *fragmentSyncer) syncFragment() error {
// syncBlock sends and receives all rows for a given block.
// Returns an error if any remote hosts are unreachable.
func (s *fragmentSyncer) syncBlock(id int) error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncBlock")
defer span.Finish()
f := s.Fragment
// Read pairs from each remote block.
@ -2283,7 +2290,7 @@ func (s *fragmentSyncer) syncBlock(id int) error {
uris = append(uris, uri)
// Only sync the standard block.
rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(context.Background(), &node.URI, f.index, f.field, f.view, f.shard, id)
rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(ctx, &node.URI, f.index, f.field, f.view, f.shard, id)
if err != nil {
return errors.Wrap(err, "getting block")
}
@ -2345,7 +2352,7 @@ func (s *fragmentSyncer) syncBlock(id int) error {
Query: buffers[k].String(),
Remote: true,
}
_, err := s.Cluster.InternalClient.QueryNode(context.Background(), uris[i], f.index, queryRequest)
_, err := s.Cluster.InternalClient.QueryNode(ctx, uris[i], f.index, queryRequest)
if err != nil {
return errors.Wrap(err, "executing")
}

11
holder.go
View file

@ -30,6 +30,7 @@ import (
"github.com/pilosa/pilosa/logger"
"github.com/pilosa/pilosa/roaring"
"github.com/pilosa/pilosa/stats"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
)
@ -720,6 +721,9 @@ func (s *holderSyncer) SyncHolder() error {
// syncIndex synchronizes index attributes with the rest of the cluster.
func (s *holderSyncer) syncIndex(index string) error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "HolderSyncer.syncIndex")
defer span.Finish()
// Retrieve index reference.
idx := s.Holder.Index(index)
if idx == nil {
@ -738,7 +742,7 @@ func (s *holderSyncer) syncIndex(index string) error {
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.ColumnAttrDiff(context.Background(), &node.URI, index, blks)
m, err := s.Cluster.InternalClient.ColumnAttrDiff(ctx, &node.URI, index, blks)
if err != nil {
return errors.Wrap(err, "getting differing blocks")
} else if len(m) == 0 {
@ -763,6 +767,9 @@ func (s *holderSyncer) syncIndex(index string) error {
// syncField synchronizes field attributes with the rest of the cluster.
func (s *holderSyncer) syncField(index, name string) error {
span, ctx := tracing.StartSpanFromContext(context.Background(), "HolderSyncer.syncField")
defer span.Finish()
// Retrieve field reference.
f := s.Holder.Field(index, name)
if f == nil {
@ -782,7 +789,7 @@ func (s *holderSyncer) syncField(index, name string) error {
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.RowAttrDiff(context.Background(), &node.URI, index, name, blks)
m, err := s.Cluster.InternalClient.RowAttrDiff(ctx, &node.URI, index, name, blks)
if err == ErrFieldNotFound {
continue // field not created remotely yet, skip
} else if err != nil {

78
http/client.go
View file

@ -29,6 +29,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/encoding/proto"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
)
@ -66,6 +67,8 @@ func NewInternalClientFromURI(defaultURI *pilosa.URI, remoteClient *http.Client)
// MaxShardByIndex returns the number of shards on a server by index.
func (c *InternalClient) MaxShardByIndex(ctx context.Context) (map[string]uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.MaxShardByIndex")
defer span.Finish()
return c.maxShardByIndex(ctx)
}
@ -100,6 +103,9 @@ func (c *InternalClient) maxShardByIndex(ctx context.Context) (map[string]uint64
// Schema returns all index and field schema information.
func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Schema")
defer span.Finish()
// Execute request against the host.
u := c.defaultURI.Path("/schema")
@ -128,6 +134,9 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
// CreateIndex creates a new index on the server.
func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilosa.IndexOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndex")
defer span.Finish()
// Encode query request.
buf, err := json.Marshal(&postIndexRequest{
Options: opt,
@ -160,6 +169,9 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
// FragmentNodes returns a list of nodes that own a shard.
func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*pilosa.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentNodes")
defer span.Finish()
// Execute request against the host.
u := uriPathToURL(c.defaultURI, "/internal/fragment/nodes")
u.RawQuery = (url.Values{"index": {index}, "shard": {strconv.FormatUint(shard, 10)}}).Encode()
@ -189,6 +201,9 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
// Nodes returns a list of all nodes.
func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Nodes")
defer span.Finish()
// Execute request against the host.
u := uriPathToURL(c.defaultURI, "/internal/nodes")
@ -217,11 +232,16 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) {
// Query executes query against the index.
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Query")
defer span.Finish()
return c.QueryNode(ctx, c.defaultURI, index, queryRequest)
}
// QueryNode executes query against the index, sending the request to the node specified.
func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "QueryNode")
defer span.Finish()
if index == "" {
return nil, pilosa.ErrIndexRequired
} else if queryRequest.Query == "" {
@ -270,6 +290,9 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s
// Import bulk imports bits for a single shard to a host.
func (c *InternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Import")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -317,6 +340,9 @@ func getCoordinatorNode(nodes []*pilosa.Node) *pilosa.Node {
// ImportK bulk imports bits specified by string keys to a host.
func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits []pilosa.Bit, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportK")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -357,6 +383,9 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, bits
}
func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options pilosa.IndexOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureIndex")
defer span.Finish()
err := c.CreateIndex(ctx, name, options)
if err == nil || errors.Cause(err) == pilosa.ErrIndexExists {
return nil
@ -365,10 +394,14 @@ func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options p
}
func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureField")
defer span.Finish()
return c.EnsureFieldWithOptions(ctx, indexName, fieldName, pilosa.FieldOptions{})
}
func (c *InternalClient) EnsureFieldWithOptions(ctx context.Context, indexName string, fieldName string, opt pilosa.FieldOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.EnsureFieldWithOptions")
defer span.Finish()
err := c.CreateFieldWithOptions(ctx, indexName, fieldName, opt)
if err == nil || errors.Cause(err) == pilosa.ErrFieldExists {
return nil
@ -404,6 +437,9 @@ func (c *InternalClient) marshalImportPayload(index, field string, shard uint64,
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
defer span.Finish()
// Create URL & HTTP request.
path := fmt.Sprintf("/index/%s/field/%s/import", index, field)
u := nodePathToURL(node, path)
@ -451,6 +487,9 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde
// ImportValue bulk imports field values for a single shard to a host.
func (c *InternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValue")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -489,6 +528,9 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
// ImportValueK bulk imports keyed field values to a host.
func (c *InternalClient) ImportValueK(ctx context.Context, index, field string, vals []pilosa.FieldValue, opts ...pilosa.ImportOption) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportValueK")
defer span.Finish()
buf, err := c.marshalImportValuePayload(index, field, 0, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
@ -547,6 +589,9 @@ func (c *InternalClient) marshalImportValuePayload(index, field string, shard ui
// ImportRoaring does fast import of raw bits in roaring format (pilosa or
// official format, see API.ImportRoaring).
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportRoaring")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -593,6 +638,9 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
// ExportCSV bulk exports data for a single shard from a host to CSV format.
func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ExportCSV")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
} else if field == "" {
@ -623,6 +671,9 @@ func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, sha
// exportNode copies a CSV export from a node to w.
func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, index, field string, shard uint64, w io.Writer) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.exportNodeCSV")
defer span.Finish()
// Create URL.
u := nodePathToURL(node, "/export")
u.RawQuery = url.Values{
@ -655,6 +706,9 @@ func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, i
}
func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
defer span.Finish()
node := &pilosa.Node{
URI: uri,
}
@ -662,6 +716,9 @@ func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field
}
func (c *InternalClient) backupShardNode(ctx context.Context, index, field string, shard uint64, node *pilosa.Node) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.backupShardNode")
defer span.Finish()
u := nodePathToURL(node, "/fragment/data")
u.RawQuery = url.Values{
"index": {index},
@ -690,11 +747,16 @@ func (c *InternalClient) backupShardNode(ctx context.Context, index, field strin
}
func (c *InternalClient) CreateField(ctx context.Context, index, field string) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateField")
defer span.Finish()
return c.CreateFieldWithOptions(ctx, index, field, pilosa.FieldOptions{})
}
// CreateField creates a new field on the server.
func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt pilosa.FieldOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateFieldWithOptions")
defer span.Finish()
if index == "" {
return pilosa.ErrIndexRequired
}
@ -749,6 +811,9 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
// FragmentBlocks returns a list of block checksums for a fragment on a host.
// Only returns blocks which contain data.
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentBlocks")
defer span.Finish()
if uri == nil {
uri = c.defaultURI
}
@ -790,6 +855,9 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
// BlockData returns row/column id pairs for a block.
func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.BlockData")
defer span.Finish()
if uri == nil {
panic("need to pass a URI to BlockData")
}
@ -835,6 +903,9 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
// ColumnAttrDiff returns data from differing blocks on a remote host.
func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pilosa.URI, index string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ColumnAttrDiff")
defer span.Finish()
if uri == nil {
uri = c.defaultURI
}
@ -872,6 +943,9 @@ func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pilosa.URI, in
// RowAttrDiff returns data from differing blocks on a remote host.
func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index, field string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RowAttrDiff")
defer span.Finish()
if uri == nil {
uri = c.defaultURI
}
@ -912,6 +986,9 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index
// SendMessage posts a message synchronously.
func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg []byte) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.SendMessage")
defer span.Finish()
u := uriPathToURL(uri, "/internal/cluster/message")
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(msg))
if err != nil {
@ -928,6 +1005,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [
// executeRequest executes the given request and checks the Response
func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, error) {
tracing.GlobalTracer.InjectHTTPHeaders(req)
resp, err := c.httpClient.Do(req)
if err != nil {
return nil, errors.Wrap(err, "executing request")

11
http/handler.go
View file

@ -36,6 +36,7 @@ import (
"github.com/gorilla/mux"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/logger"
"github.com/pilosa/pilosa/tracing"
"github.com/pkg/errors"
)
@ -220,6 +221,15 @@ func (h *Handler) queryArgValidator(next http.Handler) http.Handler {
})
}
func (h *Handler) extractTracing(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
span, ctx := tracing.GlobalTracer.ExtractHTTPHeaders(r)
defer span.Finish()
next.ServeHTTP(w, r.WithContext(ctx))
})
}
// newRouter creates a new mux http router.
func newRouter(handler *Handler) *mux.Router {
router := mux.NewRouter()
@ -260,6 +270,7 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/translate/data", handler.handleGetTranslateData).Methods("GET").Name("GetTranslateData")
router.Use(handler.queryArgValidator)
router.Use(handler.extractTracing)
return router
}

22
pilosa_internal_test.go
View file

@ -42,3 +42,25 @@ func TestValidateNameInvalid(t *testing.T) {
}
}
}
// memAttrStore represents an in-memory implementation of the AttrStore interface.
type memAttrStore struct {
store map[uint64]map[string]interface{}
}
func (s *memAttrStore) Path() string { return "" }
func (s *memAttrStore) Open() error { return nil }
func (s *memAttrStore) Close() error { return nil }
func (s *memAttrStore) Attrs(id uint64) (m map[string]interface{}, err error) { return s.store[id], nil }
func (s *memAttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
s.store[id] = m
return nil
}
func (s *memAttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
for id, v := range m {
s.store[id] = v
}
return nil
}
func (s *memAttrStore) Blocks() ([]AttrBlock, error) { return nil, nil }
func (s *memAttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) { return nil, nil }

3
server.go
View file

@ -418,6 +418,9 @@ func (s *Server) loadNodeID() string {
return nodeID
}
// NodeID returns the server's node id.
func (s *Server) NodeID() string { return s.nodeID }
// SyncData manually invokes the anti entropy process which makes sure that this
// node has the data from all replicas across the cluster.
func (s *Server) SyncData() error {

15
server/config.go
View file

@ -19,6 +19,7 @@ import (
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/toml"
"github.com/uber/jaeger-client-go"
)
// TLSConfig contains TLS configuration
@ -91,6 +92,16 @@ type Config struct {
// Pilosa's developers.
Diagnostics bool `toml:"diagnostics"`
} `toml:"metric"`
Tracing struct {
// SamplerType is the type of sampler to use.
SamplerType string `toml:"sampler-type"`
// SamplerParam is the parameter passed to the tracing sampler.
// Its meaning is dependent on the type of sampler.
SamplerParam float64 `toml:"sampler-param"`
// AgentHostPort is the host:port of the local agent.
AgentHostPort string `toml:"agent-host-port"`
} `toml:"tracing"`
}
// NewConfig returns an instance of Config with default options.
@ -133,5 +144,9 @@ func NewConfig() *Config {
c.Metric.PollInterval = toml.Duration(0 * time.Minute)
c.Metric.Diagnostics = true
// Tracing config.
c.Tracing.SamplerType = jaeger.SamplerTypeRemote
c.Tracing.SamplerParam = 0.001
return c
}

1
server/server.go
View file

@ -376,6 +376,7 @@ func (m *Command) Close() error {
eg.Go(closer.Close)
}
}
err := eg.Wait()
return errors.Wrap(err, "closing everything")
}

60
tracing/opentracing/opentracing.go Normal file
View file

@ -0,0 +1,60 @@
package opentracing
import (
"context"
"log"
"net/http"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
"github.com/pilosa/pilosa/tracing"
)
// Ensure type implements interface.
var _ tracing.Tracer = (*Tracer)(nil)
// Tracer represents a wrapper for OpenTracing that implements tracing.Tracer.
type Tracer struct {
tracer opentracing.Tracer
}
// NewTracer returns a new instance of Tracer.
func NewTracer(tracer opentracing.Tracer) *Tracer {
return &Tracer{tracer: tracer}
}
// StartSpanFromContext returns a new child span and context from a given context.
func (t *Tracer) StartSpanFromContext(ctx context.Context, operationName string) (tracing.Span, context.Context) {
var opts []opentracing.StartSpanOption
if parent := opentracing.SpanFromContext(ctx); parent != nil {
opts = append(opts, opentracing.ChildOf(parent.Context()))
}
span := t.tracer.StartSpan(operationName, opts...)
return span, opentracing.ContextWithSpan(ctx, span)
}
// InjectHTTPHeaders adds the required HTTP headers to pass context between nodes.
func (t *Tracer) InjectHTTPHeaders(r *http.Request) {
if span := opentracing.SpanFromContext(r.Context()); span != nil {
if err := t.tracer.Inject(
span.Context(),
opentracing.HTTPHeaders,
opentracing.HTTPHeadersCarrier(r.Header),
); err != nil {
log.Printf("opentracing inject error: %s", err)
}
}
}
// ExtractHTTPHeaders reads the HTTP headers to derive incoming context.
func (t *Tracer) ExtractHTTPHeaders(r *http.Request) (tracing.Span, context.Context) {
// Deserialize tracing context into request.
wireContext, _ := t.tracer.Extract(
opentracing.HTTPHeaders,
opentracing.HTTPHeadersCarrier(r.Header),
)
span := t.tracer.StartSpan("HTTP", ext.RPCServerOption(wireContext))
ctx := opentracing.ContextWithSpan(r.Context(), span)
return span, ctx
}

58
tracing/tracing.go Normal file
View file

@ -0,0 +1,58 @@
package tracing
import (
"context"
"net/http"
)
// GlobalTracer is a single, global instance of Tracer.
var GlobalTracer Tracer = NopTracer()
// StartSpanFromContext returnus a new child span and context from a given
// context using the global tracer.
func StartSpanFromContext(ctx context.Context, operationName string) (Span, context.Context) {
return GlobalTracer.StartSpanFromContext(ctx, operationName)
}
// Tracer implements a generic distributed tracing interface.
type Tracer interface {
// Returns a new child span and context from a given context.
StartSpanFromContext(ctx context.Context, operationName string) (Span, context.Context)
// Adds the required HTTP headers to pass context between nodes.
InjectHTTPHeaders(r *http.Request)
// Reads the HTTP headers to derive incoming context.
ExtractHTTPHeaders(r *http.Request) (Span, context.Context)
}
// Span represents a single span in a distributed trace.
type Span interface {
// Sets the end timestamp and finalizes Span state.
Finish()
// Adds key/value pairs to the span.
LogKV(alternatingKeyValues ...interface{})
}
// NopTracer returns a tracer that doesn't do anything.
func NopTracer() Tracer {
return &nopTracer{}
}
type nopTracer struct{}
func (t *nopTracer) StartSpanFromContext(ctx context.Context, operationName string) (Span, context.Context) {
return &nopSpan{}, ctx
}
func (t *nopTracer) InjectHTTPHeaders(r *http.Request) {}
func (t *nopTracer) ExtractHTTPHeaders(r *http.Request) (Span, context.Context) {
return &nopSpan{}, r.Context()
}
type nopSpan struct{}
func (s *nopSpan) Finish() {}
func (s *nopSpan) LogKV(alternatingKeyValues ...interface{}) {}