Merge branch 'master' into disco

This commit is contained in:
Travis 2021-02-12 20:29:41 -06:00
commit 5b237cae13
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
13 changed files with 313 additions and 46 deletions

View file

@ -11,7 +11,8 @@ BRANCH_ID := $(BRANCH)-$(GOOS)-$(GOARCH)
BUILD_TIME := $(shell date -u +%FT%T%z)
SHARD_WIDTH = 20
COMMIT := $(shell git describe --exact-match >/dev/null 2>&1 || git rev-parse --short HEAD)
LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT) -X github.com/pilosa/pilosa/v2.LatticeCommit=$(LATTICE_COMMIT)"
LDFLAGS="-X github.com/pilosa/pilosa/v2.Version=$(VERSION) -X github.com/pilosa/pilosa/v2.BuildTime=$(BUILD_TIME) -X github.com/pilosa/pilosa/v2.Variant=$(VARIANT) -X github.com/pilosa/pilosa/v2.Commit=$(COMMIT) -X github.com/pilosa/pilosa/v2.LatticeCommit=$(LATTICE_COMMIT) -X github.com/pilosa/pilosa/v2.TrialDeadline=$(TRIAL_DEADLINE)"
TRIAL_STRING = $(if $(TRIAL_DEADLINE),"-trial-$(TRIAL_DEADLINE)","")
GO_VERSION=1.14.10
RELEASE ?= 0
RELEASE_ENABLED = $(subst 0,,$(RELEASE))
@ -100,10 +101,10 @@ build:
# Create a single release build under the build directory
release-build:
$(MAKE) $(if $(DOCKER_BUILD),docker-)build FLAGS="-o build/pilosa-$(VERSION_ID)/pilosa" RELEASE=1
cp NOTICE README.md LICENSE build/pilosa-$(VERSION_ID)
tar -cvz -C build -f build/pilosa-$(VERSION_ID).tar.gz pilosa-$(VERSION_ID)/
@echo Created release build: build/pilosa-$(VERSION_ID).tar.gz
$(MAKE) $(if $(DOCKER_BUILD),docker-)build FLAGS="-o build/pilosa$(TRIAL_STRING)-$(VERSION_ID)/pilosa" RELEASE=1
cp NOTICE README.md LICENSE build/pilosa$(TRIAL_STRING)-$(VERSION_ID)
tar -cvz -C build -f build/pilosa$(TRIAL_STRING)-$(VERSION_ID).tar.gz pilosa$(TRIAL_STRING)-$(VERSION_ID)/
@echo Created release build: build/pilosa$(TRIAL_STRING)-$(VERSION_ID).tar.gz
# Error out if there are untracked changes in Git
check-clean:

37
api.go
View file

@ -172,14 +172,20 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
return QueryResponse{}, errors.Wrap(err, "validating api method")
}
if !req.Remote {
defer api.tracker.Finish(api.tracker.Start(req.Query, req.SQLQuery, api.server.nodeID, req.Index, start))
}
return api.query(ctx, req)
}
// query provides query functionality for internal use, without tracing, validation, or tracking
func (api *API) query(ctx context.Context, req *QueryRequest) (QueryResponse, error) {
q, err := pql.NewParser(strings.NewReader(req.Query)).Parse()
if err != nil {
return QueryResponse{}, errors.Wrap(err, "parsing")
}
if !req.Remote {
defer api.tracker.Finish(api.tracker.Start(req.Query, req.SQLQuery, api.server.nodeID, req.Index, start))
}
// TODO can we get rid of exec options and pass the QueryRequest directly to executor?
execOpts := &execOptions{
Remote: req.Remote,
@ -998,6 +1004,31 @@ func (api *API) Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error
return api.holder.limitedSchema()
}
// SchemaDetails returns information about each index in Pilosa including which
// fields they contain, and additional field information such as cardinality
func (api *API) SchemaDetails(ctx context.Context) ([]*IndexInfo, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
defer span.Finish()
schema := api.holder.Schema(false)
for _, index := range schema {
for _, field := range index.Fields {
q := fmt.Sprintf("Count(Distinct(field=%s))", field.Name)
req := QueryRequest{Index: index.Name, Query: q}
resp, err := api.query(ctx, &req)
if err != nil {
return schema, errors.Wrapf(err, "querying cardinality (%s/%s)", index.Name, field.Name)
}
if len(resp.Results) == 0 {
continue
}
if card, ok := resp.Results[0].(uint64); ok {
field.Cardinality = &card
}
}
}
return schema, nil
}
// ApplySchema takes the given schema and applies it across the
// cluster (if remote is false), or just to this node (if remote is
// true). This is designed for the use case of replicating a schema

View file

@ -651,6 +651,41 @@ func (e *executor) preprocessQuery(ctx context.Context, qcx *Qcx, index string,
}
}
type shardSlice []uint64
// String creates a run-length encoded representation of a slice of shard IDs (integers).
// For example, []uint64{0, 1, 3, 4, 5, 7, 8, 9, 11, 13} is represented as
// [0-1,3-5,7-9,11,13].
func (s shardSlice) String() string {
if len(s) == 0 {
// surely this is impossible
return "[]"
}
runs := make([]string, 0, len(s)/2)
start := s[0]
end := start
for n := 1; n < len(s); n++ {
if s[n] == end+1 {
end = s[n]
} else {
repr := fmt.Sprintf("%d", start)
if end > start {
repr += fmt.Sprintf("-%d", end)
}
runs = append(runs, repr)
start = s[n]
end = start
}
}
repr := fmt.Sprintf("%d", start)
if end > start {
repr += fmt.Sprintf("-%d", end)
}
runs = append(runs, repr)
return "[" + strings.Join(runs, ",") + "]"
}
// executeCall executes a call.
func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeCall")
@ -700,47 +735,47 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p
case "Sum":
statFn()
res, err := e.executeSum(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeSum %v", shards)
return res, errors.Wrapf(err, "executeSum %v", shardSlice(shards))
case "Min":
statFn()
res, err := e.executeMin(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeMin %v", shards)
return res, errors.Wrapf(err, "executeMin %v", shardSlice(shards))
case "Max":
statFn()
res, err := e.executeMax(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeMax %v", shards)
return res, errors.Wrapf(err, "executeMax %v", shardSlice(shards))
case "MinRow":
statFn()
res, err := e.executeMinRow(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeMinRow %v", shards)
return res, errors.Wrapf(err, "executeMinRow %v", shardSlice(shards))
case "MaxRow":
statFn()
res, err := e.executeMaxRow(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeMaxRow %v", shards)
return res, errors.Wrapf(err, "executeMaxRow %v", shardSlice(shards))
case "Clear":
statFn()
res, err := e.executeClearBit(ctx, qcx, index, c, opt)
return res, errors.Wrapf(err, "executeClearBit %v", shards)
return res, errors.Wrapf(err, "executeClearBit %v", shardSlice(shards))
case "ClearRow":
statFn()
res, err := e.executeClearRow(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeClearRow %v", shards)
return res, errors.Wrapf(err, "executeClearRow %v", shardSlice(shards))
case "Distinct":
statFn()
res, err := e.executeDistinct(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeDistinct %v", shards)
return res, errors.Wrapf(err, "executeDistinct %v", shardSlice(shards))
case "Store":
statFn()
res, err := e.executeSetRow(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeSetRow %v", shards)
return res, errors.Wrapf(err, "executeSetRow %v", shardSlice(shards))
case "Count":
statFn()
res, err := e.executeCount(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeCount %v", shards)
return res, errors.Wrapf(err, "executeCount %v", shardSlice(shards))
case "Set":
statFn()
res, err := e.executeSet(ctx, qcx, index, c, opt)
return res, errors.Wrapf(err, "executeSet %v", shards)
return res, errors.Wrapf(err, "executeSet %v", shardSlice(shards))
case "SetRowAttrs":
statFn()
return nil, errors.Wrap(e.executeSetRowAttrs(ctx, qcx, index, c, opt), "executeSetRowAttrs")
@ -750,50 +785,50 @@ func (e *executor) executeCall(ctx context.Context, qcx *Qcx, index string, c *p
case "TopK":
statFn()
res, err := e.executeTopK(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeTopK %v", shards)
return res, errors.Wrapf(err, "executeTopK %v", shardSlice(shards))
case "TopN":
statFn()
res, err := e.executeTopN(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeTopN %v", shards)
return res, errors.Wrapf(err, "executeTopN %v", shardSlice(shards))
case "Rows":
statFn()
res, err := e.executeRows(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeRows %v", shards)
return res, errors.Wrapf(err, "executeRows %v", shardSlice(shards))
case "Extract":
statFn()
res, err := e.executeExtract(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeExtract %v", shards)
return res, errors.Wrapf(err, "executeExtract %v", shardSlice(shards))
case "GroupBy":
statFn()
res, err := e.executeGroupBy(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeGroupBy %v", shards)
return res, errors.Wrapf(err, "executeGroupBy %v", shardSlice(shards))
case "Options":
statFn()
res, err := e.executeOptionsCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeOptionsCall %v", shards)
return res, errors.Wrapf(err, "executeOptionsCall %v", shardSlice(shards))
case "IncludesColumn":
res, err := e.executeIncludesColumnCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shards)
return res, errors.Wrapf(err, "executeIncludesColumnCall %v", shardSlice(shards))
case "FieldValue":
statFn()
res, err := e.executeFieldValueCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeFieldValueCall %v", shards)
return res, errors.Wrapf(err, "executeFieldValueCall %v", shardSlice(shards))
case "Precomputed":
res, err := e.executePrecomputedCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executePrecomputedCall %v", shards)
return res, errors.Wrapf(err, "executePrecomputedCall %v", shardSlice(shards))
case "UnionRows":
res, err := e.executeUnionRows(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeUnionRows %v", shards)
return res, errors.Wrapf(err, "executeUnionRows %v", shardSlice(shards))
case "ConstRow":
res, err := e.executeConstRow(ctx, index, c)
return res, errors.Wrapf(err, "executeConstRow %v", shards)
return res, errors.Wrapf(err, "executeConstRow %v", shardSlice(shards))
case "Limit":
res, err := e.executeLimitCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeLimitCall %v", shards)
return res, errors.Wrapf(err, "executeLimitCall %v", shardSlice(shards))
default: // e.g. "Row", "Union", "Intersect" or anything that returns a bitmap.
statFn()
res, err := e.executeBitmapCall(ctx, qcx, index, c, shards, opt)
return res, errors.Wrapf(err, "executeBitmapCall %v", shards)
return res, errors.Wrapf(err, "executeBitmapCall %v", shardSlice(shards))
}
}
@ -2864,7 +2899,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
// Get full result set.
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
if err != nil {
return nil, errors.Wrapf(err, "mapReduce shards: %v", shards)
return nil, errors.Wrapf(err, "mapReduce shards: %v", shardSlice(shards))
}
results, _ := other.([]GroupCount)
@ -5625,7 +5660,7 @@ func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch cha
// Group shards together by nodes.
m, err := e.shardsByNode(nodes, index, shards)
if err != nil {
return errors.Wrapf(err, "shards by node %v", shards)
return errors.Wrapf(err, "shards by node %v", shardSlice(shards))
}
// Execute each node in a separate goroutine.

View file

@ -1899,10 +1899,11 @@ func (p fieldSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() }
// FieldInfo represents schema information for a field.
type FieldInfo struct {
Name string `json:"name"`
CreatedAt int64 `json:"createdAt,omitempty"`
Options FieldOptions `json:"options"`
Views []*ViewInfo `json:"views,omitempty"`
Name string `json:"name"`
CreatedAt int64 `json:"createdAt,omitempty"`
Options FieldOptions `json:"options"`
Cardinality *uint64 `json:"cardinality,omitempty"`
Views []*ViewInfo `json:"views,omitempty"`
}
type fieldInfoSlice []*FieldInfo

1
go.mod
View file

@ -5,6 +5,7 @@ replace go.etcd.io/etcd => github.com/molecula/etcd v0.0.0-20210115113447-5d28bd
require (
github.com/CAFxX/gcnotifier v0.0.0-20190112062741-224a280d589d
github.com/DataDog/datadog-go v0.0.0-20180822151419-281ae9f2d895
github.com/beevik/ntp v0.3.0
github.com/benbjohnson/immutable v0.3.0
github.com/cespare/xxhash v1.1.0
github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect

2
go.sum
View file

@ -31,6 +31,8 @@ github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hC
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
github.com/beevik/ntp v0.3.0 h1:xzVrPrE4ziasFXgBVBZJDP0Wg/KpMwk2KHJ4Ba8GrDw=
github.com/beevik/ntp v0.3.0/go.mod h1:hIHWr+l3+/clUnF44zdK+CWW7fO8dR5cIylAQ76NRpg=
github.com/benbjohnson/immutable v0.3.0 h1:TVRhuZx2wG9SZ0LRdqlbs9S5BZ6Y24hJEHTCgWHZEIw=
github.com/benbjohnson/immutable v0.3.0/go.mod h1:uc6OHo6PN2++n98KHLxW8ef4W42ylHiQSENghE1ezxI=
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=

View file

@ -398,6 +398,7 @@ func newRouter(handler *Handler) http.Handler {
router.HandleFunc("/inspect", handler.handleInspect).Methods("GET").Name("Inspect")
router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST").Name("RecalculateCaches")
router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET").Name("GetSchema")
router.HandleFunc("/schema/details", handler.handleGetSchemaDetails).Methods("GET").Name("GetSchemaDetails")
router.HandleFunc("/schema", handler.handlePostSchema).Methods("POST").Name("PostSchema")
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET").Name("GetStatus")
router.HandleFunc("/transaction", handler.handlePostTransaction).Methods("POST").Name("PostTransaction")
@ -685,6 +686,24 @@ func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) {
}
}
// handleGetSchema handles GET /schema/details requests.
func (h *Handler) handleGetSchemaDetails(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
w.Header().Set("Content-Type", "application/json")
schema, err := h.api.SchemaDetails(r.Context())
if err != nil {
h.logger.Printf("error getting detailed schema: %s", err)
return
}
if err := json.NewEncoder(w).Encode(pilosa.Schema{Indexes: schema}); err != nil {
h.logger.Printf("write schema response error: %s", err)
}
}
func (h *Handler) handlePostSchema(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
remoteStr := q.Get("remote")

@ -1 +1 @@
Subproject commit 2f0302c1d124433f0e1af5ae6c3bb7e4a64ca520
Subproject commit fa773628a276e2590785a87fbc236c7e88ea6284

4
row.go
View file

@ -461,6 +461,10 @@ func (r *Row) invalidateCount() {
// Count returns the number of columns in the row.
func (r *Row) Count() uint64 {
var n uint64
if r == nil {
// Count(Distinct()) on an empty field panics here
return n
}
for i := range r.segments {
n += r.segments[i].Count()
}

View file

@ -112,6 +112,19 @@ func TestHandler_Endpoints(t *testing.T) {
})
t.Run("SchemaDetailsEmpty", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/schema/details", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
body := w.Body.String()
if body != "{\"indexes\":null}\n" {
t.Fatalf("unexpected empty schema: '%v'", body)
}
})
t.Run("PostSchema", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/schema", strings.NewReader(`{"indexes":[{"name":"blah","options":{"keys":false,"trackExistence":true},"fields":[{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}}],"shardWidth":1048576}]}`)))
@ -219,9 +232,78 @@ func TestHandler_Endpoints(t *testing.T) {
}
body := strings.TrimSpace(w.Body.String())
target := fmt.Sprintf(`{"indexes":[{"name":"i0","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}},{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}}],"shardWidth":%d},{"name":"i1","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}}],"shardWidth":%[1]d}]}`, pilosa.ShardWidth)
target := fmt.Sprintf(`{"indexes":[{"name":"i0","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}},{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}}],"shardWidth":%[1]d},{"name":"i1","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false}}],"shardWidth":%[1]d}]}`, pilosa.ShardWidth)
if body != target {
t.Fatalf("%s != %s", target, body)
t.Fatalf("\n%s\n!=\n%s", target, body)
}
})
// i2 is for SchemaDetails
i2 := hldr.MustCreateIndexIfNotExists("i2", pilosa.IndexOptions{})
tx2, err := holder.BeginTx(true, i2.Index, shard)
if err != nil {
t.Fatal(err)
}
defer tx2.Rollback()
if f, err := i2.CreateFieldIfNotExists("f0", pilosa.OptFieldTypeSet(pilosa.CacheTypeRanked, 1000)); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(tx2, 0, 0, nil); err != nil {
t.Fatal(err)
}
f, err := i2.CreateFieldIfNotExists("f1", pilosa.OptFieldTypeInt(-100, 100))
if err != nil {
t.Fatal(err)
}
for n := 0; n < 4; n++ {
if _, err := f.SetValue(tx2, uint64(n), int64(n)); err != nil {
t.Fatal(err)
}
}
f, err = i2.CreateFieldIfNotExists("f2", pilosa.OptFieldTypeDecimal(1, pql.Decimal{Value: -10}, pql.Decimal{Value: 10}))
if err != nil {
t.Fatal(err)
}
for n := 0; n < 5; n++ {
if _, err := f.SetValue(tx2, uint64(n), int64(n)); err != nil {
t.Fatal(err)
}
}
if f, err := i2.CreateFieldIfNotExists("f3", pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH"))); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(tx2, 0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i2.CreateFieldIfNotExists("f4", pilosa.OptFieldTypeMutex(pilosa.CacheTypeRanked, 5000)); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(tx2, 0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i2.CreateFieldIfNotExists("f5", pilosa.OptFieldTypeBool()); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(tx2, 0, 0, nil); err != nil {
t.Fatal(err)
}
if err := tx2.Commit(); err != nil {
t.Fatal(err)
}
t.Run("SchemaDetails", func(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/schema/details", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
}
body := strings.TrimSpace(w.Body.String())
target := fmt.Sprintf(`{"indexes":[{"name":"i0","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":0},{"name":"f1","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":1}],"shardWidth":%[1]d},{"name":"i1","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":50000,"keys":false},"cardinality":1}],"shardWidth":%[1]d},{"name":"i2","options":{"keys":false,"trackExistence":false},"fields":[{"name":"f0","options":{"type":"set","cacheType":"ranked","cacheSize":1000,"keys":false},"cardinality":1},{"name":"f1","options":{"type":"int","base":0,"bitDepth":2,"min":-100,"max":100,"keys":false,"foreignIndex":""},"cardinality":4},{"name":"f2","options":{"type":"decimal","base":0,"scale":1,"bitDepth":3,"min":-10,"max":10,"keys":false},"cardinality":5},{"name":"f3","options":{"type":"time","timeQuantum":"YMDH","keys":false,"noStandardView":false},"cardinality":1},{"name":"f4","options":{"type":"mutex","cacheType":"ranked","cacheSize":5000,"keys":false},"cardinality":1},{"name":"f5","options":{"type":"bool"},"cardinality":1}],"shardWidth":%[1]d}]}`, pilosa.ShardWidth)
if body != target {
t.Fatalf("\n%s\n!=\n%s", target, body)
}
})
@ -402,13 +484,13 @@ func TestHandler_Endpoints(t *testing.T) {
for _, nodeUsage := range nodeUsages {
numIndexes := len(nodeUsage.Disk.IndexUsage)
if nodeUsage.Disk.TotalUse < 75000 || nodeUsage.Disk.TotalUse > 300000 {
if nodeUsage.Disk.TotalUse < 75000 || nodeUsage.Disk.TotalUse > 500000 {
// Usage measurements are not consistent between machines, or
// over time, as features and implementations change, so checking
// for a range of sizes may be most useful way to test the details of this.
t.Fatalf("expected 75k < total < 300k, got %d", nodeUsage.Disk.TotalUse)
t.Fatalf("expected 75k < total < 500k, got %d", nodeUsage.Disk.TotalUse)
}
if numIndexes != 2 {
if numIndexes != 3 {
t.Fatalf("wrong length index usage list: expected %d, got %d", 2, numIndexes)
}
numFields := len(nodeUsage.Disk.IndexUsage["i1"].Fields)
@ -487,7 +569,7 @@ func TestHandler_Endpoints(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/internal/shards/max", nil))
if w.Code != gohttp.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"standard":{"i0":3,"i1":0}}`+"\n" {
} else if body := w.Body.String(); body != `{"standard":{"i0":3,"i1":0,"i2":0}}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
})

View file

@ -276,6 +276,8 @@ func (m *Command) SetupServer() error {
m.logger.Printf("%s", pilosa.VersionInfo())
handleTrialDeadline(m.logger)
// validateAddrs sets the appropriate values for Bind and Advertise
// based on the inputs. It is not responsible for applying defaults, although
// it does provide a non-zero port (10101) in the case where no port is specified.

86
server/trial.go Normal file
View file

@ -0,0 +1,86 @@
// Copyright 2021 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Package server contains the `pilosa server` subcommand which runs Pilosa
// itself. The purpose of this package is to define an easily tested Command
// object which handles interpreting configuration and setting up all the
// objects that Pilosa needs.
package server
import (
"os"
"time"
"github.com/beevik/ntp"
"github.com/pilosa/pilosa/v2"
)
// handleTrialDeadline checks to see if this is a trial version of Molecula that expires at some point.
// If it is, we contact an NTP server to get the current time and compare that to the trial deadline.
// We launch two goroutines, one which reminds via a log message how much time is left in the trial,
// and one which causes the process to exit once the trial is over.
func handleTrialDeadline(logger loggerLogger) {
if pilosa.TrialDeadline != "" {
startTime, err := ntpServerTime(4, logger)
if err != nil {
logger.Printf("reading ntp server time %v", err)
os.Exit(1)
}
endTime, err := time.Parse("2006-01-02", pilosa.TrialDeadline)
if err != nil {
logger.Printf("parsing trial deadline: %v", err)
os.Exit(1)
}
maxDuration := endTime.Sub(startTime)
go expireAfter(maxDuration, logger)
go dailyCheck(maxDuration, logger)
}
}
const trialCheckInterval = 24 * time.Hour
// dailyCheck runs in the background while a trial version of Molecula is being run, displaying daily reminders of the remaining days
func dailyCheck(maxDuration time.Duration, logger loggerLogger) {
startTime := time.Now() // we get a new start time here to ensure that it has a monotonic clock
remaining := maxDuration - time.Since(startTime)
logger.Printf("Current time remaining in trial: %d days %v", remaining/(time.Hour*24), remaining%(time.Hour*24))
ticker := time.NewTicker(trialCheckInterval)
for range ticker.C {
remaining := maxDuration - time.Since(startTime)
logger.Printf("Current time remaining in trial: %d days %v", remaining/(time.Hour*24), remaining%(time.Hour*24))
}
}
const ntpURL = "0.beevik-ntp.pool.ntp.org"
const ntpRetryDelay = 100 * time.Millisecond
// ntpServerTime attempts to reach ntp servers with delays between each attempt, returning the time value of the first connected server
func ntpServerTime(retries int, logger loggerLogger) (time.Time, error) {
t, err := ntp.Time(ntpURL)
if err != nil && retries <= 0 {
return t, err
}
if err != nil {
time.Sleep(ntpRetryDelay)
return ntpServerTime(retries-1, logger)
}
return t, nil
}
func expireAfter(maxDuration time.Duration, logger loggerLogger) {
time.Sleep(maxDuration)
logger.Printf("Trial edition of Molecula has expired, exiting now!")
os.Exit(1)
}

View file

@ -25,6 +25,7 @@ var Variant string
var BuildTime string
var LatticeCommit string
var GoVersion string = runtime.Version()
var TrialDeadline string
func VersionInfo() string {
var prefix string
@ -53,7 +54,9 @@ func VersionInfo() string {
suffix += " (" + buildTime + ")"
}
suffix += " " + GoVersion
if TrialDeadline != "" {
suffix += " limited time trial ends on: " + TrialDeadline
}
return prefix + "Pilosa" + suffix
}