From 000f708c40786e819acac095a281e8b5f97cb221 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Wed, 24 Mar 2021 18:09:01 +0100 Subject: [PATCH] Address PR comments + some docs --- client/README.md | 85 ++++++++++++++ client/client.go | 4 +- client/client_test.go | 43 ++++++-- client/doc.go | 10 +- client/docs/data-model-queries.md | 152 +++++++++++++++++++++++++ client/docs/server-interaction.md | 178 ++++++++++++++++++++++++++++++ client/docs/tracing.md | 111 +++++++++++++++++++ net/uri.go | 6 - 8 files changed, 570 insertions(+), 19 deletions(-) create mode 100644 client/README.md create mode 100644 client/docs/data-model-queries.md create mode 100644 client/docs/server-interaction.md create mode 100644 client/docs/tracing.md diff --git a/client/README.md b/client/README.md new file mode 100644 index 000000000..9af16df33 --- /dev/null +++ b/client/README.md @@ -0,0 +1,85 @@ +# Go Client for Pilosa + +Go client for Pilosa high performance distributed index. + +## Usage + +If you have the pilosa repo in your `GOPATH`, +you can import the library in your code using: + +```go +import "github.com/pilosa/pilosa/v2/client" +``` + + +### Quick overview + +Assuming [Pilosa](https://github.com/pilosa/pilosa) server is running at `localhost:10101` (the default): + +```go +package main + +import ( + "fmt" + + "github.com/pilosa/pilosa/v2/client" +) + +func main() { + // Create the default client + cli := client.DefaultClient() + + // Retrieve the schema + schema, err := cli.Schema() + + // Create an Index object + myindex := schema.Index("myindex") + + // Create a Field object + myfield := myindex.Field("myfield") + + // make sure the index and the field exists on the server + err := cli.SyncSchema(schema) + + // Send a Set query. If err is non-nil, response will be nil. + response, err := cli.Query(myfield.Set(5, 42)) + + // Send a Row query. If err is non-nil, response will be nil. + response, err = cli.Query(myfield.Row(5)) + + // Get the result + result := response.Result() + // Act on the result + if result != nil { + columns := result.Row().Columns + fmt.Println("Got columns: ", columns) + } + + // You can batch queries to improve throughput + response, err = cli.Query(myindex.BatchQuery( + myfield.Row(5), + myfield.Row(10))) + if err != nil { + fmt.Println(err) + } + + for _, result := range response.Results() { + // Act on the result + fmt.Println(result.Row().Columns) + } +} +``` + +## Documentation + +### Data Model and Queries + +See: [Data Model and Queries](docs/data-model-queries.md) + +### Executing Queries + +See: [Server Interaction](docs/server-interaction.md) + +### Other Documentation + +* [Tracing](docs/tracing.md) \ No newline at end of file diff --git a/client/client.go b/client/client.go index f3542697e..e4e8a40c3 100644 --- a/client/client.go +++ b/client/client.go @@ -873,7 +873,9 @@ func (c *Client) host(usePrimary bool) (*pnet.URI, error) { c.primaryLock.Unlock() return nil, errors.Wrap(err, "fetching primary node") } - host = pnet.URIFromAddress(fmt.Sprintf("%s://%s:%d", node.Scheme, node.Host, node.Port)) + if host, err = pnet.NewURIFromAddress(fmt.Sprintf("%s://%s:%d", node.Scheme, node.Host, node.Port)); err != nil { + return nil, errors.Wrap(err, "parsing primary node URL") + } } else { host = c.primaryURI } diff --git a/client/client_test.go b/client/client_test.go index 22c615bb4..0833b7739 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -79,7 +79,10 @@ func TestNewClient(t *testing.T) { if err != nil { t.Fatal(err) } - targetURI := pnet.URIFromAddress(":9999") + targetURI, err := pnet.NewURIFromAddress(":9999") + if err != nil { + t.Fatal(err) + } if !reflect.DeepEqual(targetURI, client.manualServerURI) { t.Fatalf("%v != %v", targetURI, client.manualServerURI) } @@ -95,7 +98,13 @@ func TestNewClient(t *testing.T) { if err != nil { t.Fatal(err) } - target := []*pnet.URI{pnet.URIFromAddress(":9999")} + + targetURI, err = pnet.NewURIFromAddress(":9999") + if err != nil { + t.Fatal(err) + } + + target := []*pnet.URI{targetURI} if !reflect.DeepEqual(target, client.cluster.hosts) { t.Fatalf("%v != %v", target, client.cluster.hosts) } @@ -107,20 +116,30 @@ func TestNewClient(t *testing.T) { t.Fatalf("%v != %v", target, client.cluster.hosts) } - client, err = NewClient([]*pnet.URI{pnet.URIFromAddress(":9999"), pnet.URIFromAddress(":8888")}) + targetURI1, err := pnet.NewURIFromAddress(":8888") if err != nil { t.Fatal(err) } - target = []*pnet.URI{pnet.URIFromAddress(":9999"), pnet.URIFromAddress(":8888")} + targetURI2, err := pnet.NewURIFromAddress(":9999") + if err != nil { + t.Fatal(err) + } + + client, err = NewClient([]*pnet.URI{targetURI1, targetURI2}) + if err != nil { + t.Fatal(err) + } + + target = []*pnet.URI{targetURI1, targetURI2} if !reflect.DeepEqual(target, client.cluster.hosts) { t.Fatalf("%v != %v", target, client.cluster.hosts) } - client, err = NewClient([]*pnet.URI{pnet.URIFromAddress(":9999")}) + client, err = NewClient([]*pnet.URI{targetURI}) if err != nil { t.Fatal(err) } - target = []*pnet.URI{pnet.URIFromAddress(":9999")} + target = []*pnet.URI{targetURI} if !reflect.DeepEqual(target, client.cluster.hosts) { t.Fatalf("%v != %v", target, client.cluster.hosts) } @@ -166,7 +185,17 @@ func TestNewClientManualAddressWithMultipleURIs(t *testing.T) { if err != ErrSingleServerAddressRequired { t.Fatalf("%v != %v", ErrSingleServerAddressRequired, err) } - _, err = NewClient([]*pnet.URI{pnet.URIFromAddress(":9000"), pnet.URIFromAddress(":5000")}, OptClientManualServerAddress(true)) + + targetURI1, err := pnet.NewURIFromAddress(":9000") + if err != nil { + t.Fatal(err) + } + targetURI2, err := pnet.NewURIFromAddress(":5000") + if err != nil { + t.Fatal(err) + } + + _, err = NewClient([]*pnet.URI{targetURI1, targetURI2}, OptClientManualServerAddress(true)) if err != ErrSingleServerAddressRequired { t.Fatalf("%v != %v", ErrSingleServerAddressRequired, err) } diff --git a/client/doc.go b/client/doc.go index 43ab67b93..ea7153e01 100644 --- a/client/doc.go +++ b/client/doc.go @@ -16,7 +16,7 @@ // generally administration, testing, and debugging tools. /* -Package pilosa enables querying a Pilosa server. +Package client enables querying a Pilosa server. This client uses Pilosa's http+protobuf API. @@ -28,10 +28,10 @@ Usage: ) // Create a Client instance - client := client.DefaultClient() + cli := client.DefaultClient() // Create a Schema instance - schema, err := client.Schema() + schema, err := cli.Schema() if err != nil { panic(err) } @@ -49,13 +49,13 @@ Usage: } // Sync the schema with the server-side, so non-existing indexes/fields are created on the server-side. - err = client.SyncSchema(schema) + err = cli.SyncSchema(schema) if err != nil { panic(err) } // Execute a query - response, err := client.Query(stargazer.Row(5)) + response, err := cli.Query(stargazer.Row(5)) if err != nil { panic(err) } diff --git a/client/docs/data-model-queries.md b/client/docs/data-model-queries.md new file mode 100644 index 000000000..9e88f7781 --- /dev/null +++ b/client/docs/data-model-queries.md @@ -0,0 +1,152 @@ +# Data Model and Queries + +## Indexes and Fields + +*Index* and *field*s are the main data models of Pilosa. You can check the [Pilosa documentation](https://www.pilosa.com/docs/latest/data-model/) for more detail about the data model. + +The `schema.Index` function is used to create an index instance. Note that this does not create an index on the server; the index object simply defines the schema. + +```go +schema := client.NewSchema() +repository := schema.Index("repository") +``` + +You can pass options while creating index instances: +```go +repository := schema.Index("repository", pilosa.OptIndexKeys(true)) +``` + +Field definitions are created with a call to the `Field` function of an index: + +```go +stargazer := repository.Field("stargazer") +``` + +You can pass options to fields: + +```go +stargazer := repository.Field("stargazer", pilosa.OptFieldTypeTime(TimeQuantumYearMonthDay)) +``` + +In case the schema already exists on the server, you can retrieve that instead of creating the schema: +```go +cli := client.DefaultClient() +schema, err := cli.Schema() +if err != nil { + // act on the error +} +repository := schema.Index("repository") +``` + +## Queries + +Once you have indexes and field definitions, you can create queries for them. Some of the queries work on the columns; corresponding methods are attached to the index. Other queries work on rows with related methods attached to fields. + +For instance, `Row` queries work on rows; use a `Field` object to create those queries: + +```go +rowQuery := stargazer.Row(1) // corresponds to PQL: Row(stargazer=1) +``` + +`Union` queries work on columns; use the index to create them: + +```go +query := repository.Union(rowQuery1, rowQuery2) +``` + +In order to increase throughput, you may want to batch queries sent to the Pilosa server. The `index.BatchQuery` function is used for that purpose: + +```go +query := repository.BatchQuery( + stargazer.Row(1), + repository.Union(stargazer.Row(100), stargazer.Row(5))) +``` + +The recommended way of creating query instances is using dedicated functions attached to index and field objects, but sometimes it would be desirable to send raw queries to Pilosa. You can use `index.RawQuery` method for that. Note that query string is not validated before sending to the server: + +```go +query := repository.RawQuery("Row(stargazer=5)") +``` + +Raw queries are only sent to the coordinator node of a Pilosa cluster, so currently there's a possible performance hit using them instead of ORM functions attached to index or field instances. + +This client supports [range queries using bit sliced indexes (BSI)](https://www.pilosa.com/docs/latest/query-language/#range-bsi). Read the [Range Encoded Bitmaps](https://www.pilosa.com/blog/range-encoded-bitmaps/) blog post for more information about the BSI implementation of range encoding in Pilosa. + +In order to use BSI range queries, an integer field should be created. The field should have its minimum and maximum set. Here's how you would do that: +```go +index := schema.Index("animals") +captivity := index.Field("captivity", pilosa.OptFieldTypeInt(0, 956)) +``` + +If the field with the necessary field already exists on the server, you don't need to create the field instance, `cli.SyncSchema(schema)` would load that to `schema`. You can then add some data: +```go +// Add the captivity values to the field. +data := []int{3, 392, 47, 956, 219, 14, 47, 504, 21, 0, 123, 318} +query := index.BatchQuery() +for i, x := range data { + column := uint64(i + 1) + query.Add(captivity.SetIntValue(column, x)) +} +cli.Query(query) +``` + +Let's write a range query: +```go +// Query for all animals with more than 100 specimens +response, _ := cli.Query(captivity.GT(100)) +fmt.Println(response.Result().Row().Columns) + +// Query for the total number of animals in captivity +response, _ = cli.Query(captivity.Sum(nil)) +fmt.Println(response.Result().Value()) +``` + +If you pass a row query to `Sum` as a filter, then only the columns matching the filter will be considered in the `Sum` calculation: +```go +// Let's run a few set queries first +cli.Query(index.BatchQuery( + field.Set(42, 1), + field.Set(42, 6))) +// Query for the total number of animals in captivity where row 42 is set +response, _ = cli.Query(captivity.Sum(field.Row(42))) +fmt.Println(response.Result().Value()) +``` + +See the functions further below for the list of functions that can be used with a `Field`. + +Please check [Pilosa documentation](https://www.pilosa.com/docs) for PQL details. Here is a list of methods corresponding to PQL calls: + +Index: + +* `Union(rows *PQLRowQuery...) *PQLRowQuery` +* `Intersect(rows *PQLRowQuery...) *PQLRowQuery` +* `Difference(rows *PQLRowQuery...) *PQLRowQuery` +* `Xor(rows ...*PQLRowQuery) *PQLRowQuery` +* `Not(row) *PQLRowQuery` +* `Count(row *PQLRowQuery) *PQLBaseQuery` +* `SetColumnAttrs(columnID uint64, attrs map[string]interface{}) *PQLBaseQuery` +* `Options(row *PQLRowQuery, opts ...OptionsOption) *PQLBaseQuery` + +Field: + +* `Row(rowID uint64) *PQLRowQuery` +* `Set(rowID uint64, columnID uint64) *PQLBaseQuery` +* `SetTimestamp(rowID uint64, columnID uint64, timestamp time.Time) *PQLBaseQuery` +* `Clear(rowID uint64, columnID uint64) *PQLBaseQuery` +* `TopN(n uint64) *PQLRowQuery` +* `RowTopN(n uint64, row *PQLRowQuery) *PQLRowQuery` +* `FilterFieldTopN(n uint64, row *PQLRowQuery, field string, values ...interface{}) *PQLRowQuery` +* `Range(rowID uint64, start time.Time, end time.Time) *PQLRowQuery` +* `RowRange(rowID uint64, start time.Time, end time.Time) *PQLRowQuery` +* `SetRowAttrs(rowID uint64, attrs map[string]interface{}) *PQLBaseQuery` +* `ClearRow(rowIDOrKey interface{}) *PQLBaseQuery` +* `Store(row *PQLRowQuery, rowIDOrKey interface{}) *PQLBaseQuery` +* `LT(n int) *PQLRowQuery` +* `LTE(n int) *PQLRowQuery` +* `GT(n int) *PQLRowQuery` +* `GTE(n int) *PQLRowQuery` +* `Between(a int, b int) *PQLRowQuery` +* `Sum(row *PQLRowQuery) *PQLBaseQuery` +* `Min(row *PQLRowQuery) *PQLBaseQuery` +* `Max(row *PQLRowQuery) *PQLBaseQuery` +* `SetIntValue(columnID uint64, value int) *PQLBaseQuery` diff --git a/client/docs/server-interaction.md b/client/docs/server-interaction.md new file mode 100644 index 000000000..a713c30c3 --- /dev/null +++ b/client/docs/server-interaction.md @@ -0,0 +1,178 @@ +# Server Interaction + +## Pilosa URI + +A Pilosa URI has the `${SCHEME}://${HOST}:${PORT}` format: +* **Scheme**: Protocol of the URI. Default: `http`. +* **Host**: Hostname or ipv4/ipv6 IP address. Default: localhost. +* **Port**: Port number. Default: `10101`. + +All parts of the URI are optional, but at least one of them must be specified. The following are equivalent: + +* `http://localhost:10101` +* `http://localhost` +* `http://:10101` +* `localhost:10101` +* `localhost` +* `:10101` + +A Pilosa URI is represented by the `github.com/pilosa/pilosa/v2/net URI` struct. Below are a few ways to create `URI` objects: + +```go +import pnet "github.com/pilosa/pilosa/v2/net" + +// create the default URI: http://localhost:10101 +uri1 := pnet.DefaultURI() + +// create a URI from string address +uri2, err := pnet.NewURIFromAddress("index1.pilosa.com:20202"); + +// create a URI with the given host and port +uri3, err := pnet.NewURIFromHostPort("index1.pilosa.com", 20202); +``` + +## Pilosa Client + +In order to interact with a Pilosa server, an instance of `client.Client` should be created. The client is thread-safe and uses a pool of connections to the server, so we recommend creating a single instance of the client and sharing it when necessary. + +If the Pilosa server is running at the default address (`http://localhost:10101`) you can create the client with default options using: + +```go +import "github.com/pilosa/pilosa/v2/client" + +cli := client.DefaultClient() +``` + +To use a custom server address, use the `NewClient` function: + +```go +uri, err := pnet.NewURIFromAddress("http://index1.pilosa.com:15000") +if err != nil { + // Act on the error +} +cli, err := client.NewClient(uri) +``` + +Equivalently: +```go +cli, err := client.NewClient("http://index1.pilosa.com:15000") +``` + +If you are running a cluster of Pilosa servers, you can create a `Cluster` struct that keeps addresses of those servers: + +```go +uri1, err := pnet.NewURIFromAddress(":10101") +uri2, err := pnet.NewURIFromAddress(":10110") +uri3, err := pnet.NewURIFromAddress(":10111") +cluster := client.NewClusterWithHost(uri1, uri2, uri3) + +// Create a client with the cluster +cli, err := client.NewClient(cluster) +``` + +That is equivalent to: +```go +cli, err := client.NewClient([]string{":10101", ":10110", ":10111"}) + +``` + +It is possible to customize the behaviour of the underlying HTTP client by passing `ClientOption` structs to the `NewClient` function: + +```go +cli, err := client.NewClient(cluster, + client.OptClientConnectTimeout(1000), // if can't connect in a second, close the connection + client.OptClientSocketTimeout(10000), // if no response received in 10 seconds, close the connection + client.OptClientPoolSizePerRoute(3), // number of connections in the pool per host + client.OptClientTotalPoolSize(10)) // number of total connections in the pool +``` + +Once you create a client, you can create indexes, fields or start sending queries. + +Here is how you would create a index and field: + +```go +// materialize repository index definition and stargazer field definition initialized before +err := cli.SyncSchema(schema) +``` + +You can send queries to a Pilosa server using the `Query` function of the `Client` struct: + +```go +response, err := cli.Query(field.Row(5)); +``` + +`Query` accepts zero or more options: + +```go +response, err := cli.Query(field.Row(5), pilosa.ColumnAttrs(true), pilosa.ExcludeColumns(true)) +``` + +## Server Response + +When a query is sent to a Pilosa server, the server either fulfills the query or sends an error message. In the case of an error, a `pilosa.Error` struct is returned, otherwise a `QueryResponse` struct is returned. + +A `QueryResponse` struct may contain zero or more results of `QueryResult` type. You can access all results using the `Results` function of `QueryResponse` (which returns a list of `QueryResult` objects), or you can use the `Result` method (which returns either the first result or `nil` if there are no results): + +```go +response, err := cli.Query(field.Row(5)) +if err != nil { + // Act on the error +} + +// check that there's a result and act on it +result := response.Result() +if result != nil { + // Act on the result +} + +// iterate over all results +for _, result := range response.Results() { + // Act on the result +} +``` + +Similarly, a `QueryResponse` struct may include a number of column attributes if `ColumnAttrs` query option was set to `true`: + +```go +var column *pilosa.ColumnItem + +// iterate over all columns +for _, column = range response.ColumnAttrs() { + // Act on the column item +} +``` + +`QueryResult` objects contain: + +* `Row()` function to retrieve a row result, +* `CountItems()` function to retrieve column count per row ID entries returned from `TopN` queries, +* `Count()` function to retrieve the number of rows per the given row ID returned from `Count` queries. +* `Value()` function to retrieve the result of `Min`, `Max` or `Sum` queries. +* `Changed()` function returns whether a `Set` or `Clear` query changed a column. + +```go +row := result.Row() +columns := row.Columns +attributes := row.Attributes + +countItems := result.CountItems() + +count := result.Count() + +value := result.Value() + +changed := result.Changed() +``` + +## SSL/TLS + +Make sure the Pilosa server runs on a TLS address. [How To Set Up a Secure Cluster](https://www.pilosa.com/docs/latest/tutorials/#how-to-set-up-a-secure-cluster) tutorial explains how to do that. + +In order to enable TLS support on the client side, the scheme of the address should be explicitly specified as `https`, e.g.: `https://01.pilosa.local:10501` + +This client library uses the `net/http` module of Go standard library. You can pass a [tls.Config](https://golang.org/pkg/crypto/tls/#Config) struct in a `pilosa.TLSConfig` option to the client. If the Pilosa server is using a certificate from a recognized authority, you can use the defaults. + +If you are using a self signed certificate, just pass `pilosa.TLSConfig(&tls.Config{InsecureSkipVerify: true})` to `pilosa.NewClient` function: +```go +client, _ := pilosa.NewClient("https://01.pilosa.local:10501", pilosa.TLSConfig(&tls.Config{InsecureSkipVerify: true})) +``` diff --git a/client/docs/tracing.md b/client/docs/tracing.md new file mode 100644 index 000000000..868e374df --- /dev/null +++ b/client/docs/tracing.md @@ -0,0 +1,111 @@ +# Tracing + +Pilosa client supports distributed tracing via the [OpenTracing](https://opentracing.io/) API. + +In order to use a tracer with Go-Pilosa, you should: +1. Create the tracer, +2. Pass the `OptClientOption(tracer)` to `NewClient`. + +In this document, we will be using the [Jaeger](https://www.jaegertracing.io) tracer, but OpenTracing has support for [other tracing systems](https://opentracing.io/docs/supported-tracers/). + +## Running the Pilosa Server + +Let's run a temporary Pilosa container: + + $ docker run -it --rm -p 10101:10101 pilosa/pilosa:v1.2.0 + +Check that you can access Pilosa: + + $ curl localhost:10101 + Welcome. Pilosa is running. Visit https://www.pilosa.com/docs/ for more information. + +## Running the Jaeger Server + +Let's run a Jaeger Server container: + + $ docker run -it --rm -p 5775:5775/udp -p 16686:16686 jaegertracing/all-in-one:latest + ...Jaeger UI... + +## Writing the Sample Code + +The sample code depdends on the Jaeger Go client, so let's install it first: + + $ go get -u github.com/uber/jaeger-client-go/ + +Save the following sample code as `gopilosa-tracing.go`: +```go +package main + +import ( + "log" + "time" + + "github.com/pilosa/pilosa/v2/client" + "github.com/uber/jaeger-client-go" + "github.com/uber/jaeger-client-go/config" +) + +func main() { + // Create the tracer. + cfg := config.Configuration{ + Sampler: &config.SamplerConfig{ + Type: "const", + Param: 1, + }, + Reporter: &config.ReporterConfig{ + LogSpans: true, + BufferFlushInterval: 1 * time.Second, + // Jaeger Server address + LocalAgentHostPort: "127.0.0.1:5775", + }, + } + tracer, closer, err := cfg.New( + "go_pilosa_test", + config.Logger(jaeger.StdLogger), + ) + + // Don't forget to close the tracer. + defer closer.Close() + + // Create the client, and pass the tracer. + cli, err := client.NewClient(":10101", pilosa.OptClientTracer(tracer)) + if err != nil { + log.Fatal(err) + } + + // Read the schema from the server. + // This should create a trace on the Jaeger server. + schema, err := cli.Schema() + if err != nil { + log.Fatal(err) + } + + // Create and sync the sample schema. + // This should create a trace on the Jaeger server. + myIndex := schema.Index("my-index") + myField := myIndex.Field("my-field") + err = cli.SyncSchema(schema) + if err != nil { + log.Fatal(err) + } + + // Run a query on Pilosa. + // This should create a trace on the Jaeger server. + _, err = cli.Query(myField.Set(1, 1000)) + if err != nil { + log.Fatal(err) + } +} +``` + +## Checking the Tracing Data + +Run the sample code: + + $ go run gopilosa-tracing.go + + +* Open http://localhost:16686 in your web browser to visit Jaeger UI. +* Click on the *Search* tab and select `go_pilosa_test` in the *Service* dropdown on the right. +* Click on *Find Traces* button at the bottom left. +* You should see a couple of traces, such as: `Client.Query`, `Client.CreateField`, `Client.Schema`, etc. diff --git a/net/uri.go b/net/uri.go index e849e77bd..c29850036 100644 --- a/net/uri.go +++ b/net/uri.go @@ -97,12 +97,6 @@ func NewURIFromAddress(address string) (*URI, error) { return parseAddress(address) } -// URIFromAddress creates a URI from the given address. -func URIFromAddress(host string) *URI { - uri, _ := NewURIFromAddress(host) - return uri -} - // SetScheme sets the scheme of this URI. func (u *URI) SetScheme(scheme string) error { m := schemeRegexp.FindStringSubmatch(scheme)