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

Add distributed tracing.

Browse source
This commit is contained in:
Ben Johnson 2018-10-09 08:28:21 -06:00 • committed by Cody Soyland
parent ef21492ae0
commit 8e49332b25
17 changed files with 728 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")
},
}

4
ctl/server.go
View file

@ -68,4 +68,8 @@ 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.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.")
}

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{}) {}