From 50794bf63be39fd45807255f58c3b340427ee873 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Sun, 24 Jun 2018 23:59:35 -0500 Subject: [PATCH 1/6] move fieldOptions unmarshal to the handler validate fieldOptions in http package --- api.go | 14 +++- client.go | 8 +- cmd/import.go | 7 +- ctl/import.go | 4 +- field.go | 107 +++++++++++++++++++------ field_test.go | 15 ++-- fragment.go | 4 +- fragment_internal_test.go | 2 +- http/client.go | 9 ++- http/handler.go | 144 ++++++++++++++++++++++++---------- http/handler_internal_test.go | 113 +++++++++++++++++++++++--- index.go | 2 +- server/cluster_test.go | 10 +-- server/server_test.go | 13 +-- server_test.go | 3 +- test/field.go | 12 +-- view.go | 2 +- view_internal_test.go | 2 +- 18 files changed, 346 insertions(+), 125 deletions(-) diff --git a/api.go b/api.go index 98fd5d717..dd0d5fca4 100644 --- a/api.go +++ b/api.go @@ -244,11 +244,19 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error { } // CreateField makes the named field in the named index with the given options. -func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, options FieldOptions) (*Field, error) { +func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, opts ...FieldOption) (*Field, error) { if err := api.validate(apiCreateField); err != nil { return nil, errors.Wrap(err, "validating api method") } + fo := FieldOptions{} + for _, opt := range opts { + err := opt(&fo) + if err != nil { + return nil, errors.Wrap(err, "applying option") + } + } + // Find index. index := api.Holder.Index(indexName) if index == nil { @@ -256,7 +264,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str } // Create field. - field, err := index.CreateField(fieldName, options) + field, err := index.CreateField(fieldName, fo) if err != nil { return nil, errors.Wrap(err, "creating field") } @@ -266,7 +274,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str &internal.CreateFieldMessage{ Index: indexName, Field: fieldName, - Meta: options.Encode(), + Meta: fo.Encode(), }) if err != nil { api.server.logger.Printf("problem sending CreateField message: %s", err) diff --git a/client.go b/client.go index b5dbe809a..292a71ef6 100644 --- a/client.go +++ b/client.go @@ -41,10 +41,10 @@ type InternalClient interface { Import(ctx context.Context, index, field string, slice uint64, bits []Bit) error ImportK(ctx context.Context, index, field string, bits []Bit) error EnsureIndex(ctx context.Context, name string, options IndexOptions) error - EnsureField(ctx context.Context, indexName string, fieldName string, options FieldOptions) error + EnsureField(ctx context.Context, indexName string, fieldName string) error ImportValue(ctx context.Context, index, field string, slice uint64, vals []FieldValue) error ExportCSV(ctx context.Context, index, field string, slice uint64, w io.Writer) error - CreateField(ctx context.Context, index, field string, opt FieldOptions) error + CreateField(ctx context.Context, index, field string) error FragmentBlocks(ctx context.Context, uri *URI, index, field string, slice uint64) ([]FragmentBlock, error) BlockData(ctx context.Context, uri *URI, index, field string, slice uint64, block int) ([]uint64, []uint64, error) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) @@ -108,7 +108,7 @@ func (n *NopInternalClient) ImportK(ctx context.Context, index, field string, bi func (n *NopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error { return nil } -func (n *NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string, options FieldOptions) error { +func (n *NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error { return nil } func (n *NopInternalClient) ImportValue(ctx context.Context, index, field string, slice uint64, vals []FieldValue) error { @@ -117,7 +117,7 @@ func (n *NopInternalClient) ImportValue(ctx context.Context, index, field string func (n *NopInternalClient) ExportCSV(ctx context.Context, index, field string, slice uint64, w io.Writer) error { return nil } -func (n *NopInternalClient) CreateField(ctx context.Context, index, field string, opt FieldOptions) error { +func (n *NopInternalClient) CreateField(ctx context.Context, index, field string) error { return nil } func (n *NopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field string, slice uint64) ([]FragmentBlock, error) { diff --git a/cmd/import.go b/cmd/import.go index db5cebf8d..5a27b4055 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -20,7 +20,6 @@ import ( "github.com/spf13/cobra" - "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/ctl" ) @@ -59,9 +58,9 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.") flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.") flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.") - flags.Var(&Importer.FieldOptions.TimeQuantum, "field-time-quantum", "Time quantum for the field") - flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Cache type for the field; valid values: none, lru, ranked") - flags.Uint32Var(&Importer.FieldOptions.CacheSize, "field-cache-size", 50000, "Cache size for the field") + //flags.Var(&Importer.FieldOptions.TimeQuantum, "field-time-quantum", "Time quantum for the field") + //flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Cache type for the field; valid values: none, lru, ranked") + //flags.Uint32Var(&Importer.FieldOptions.CacheSize, "field-cache-size", 50000, "Cache size for the field") ctl.SetTLSConfig(flags, &Importer.TLS.CertificatePath, &Importer.TLS.CertificateKeyPath, &Importer.TLS.SkipVerify) return importCmd diff --git a/ctl/import.go b/ctl/import.go index c4d65f232..ad4befbfd 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -42,7 +42,7 @@ type ImportCommand struct { // Options for index & field to be created if they don't exist IndexOptions pilosa.IndexOptions - FieldOptions pilosa.FieldOptions + //FieldOptions pilosa.FieldOptions // CreateSchema ensures the schema exists before import CreateSchema bool @@ -135,7 +135,7 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error { if err != nil { return fmt.Errorf("Error Creating Index: %s", err) } - err = cmd.Client.EnsureField(ctx, cmd.Index, cmd.Field, cmd.FieldOptions) + err = cmd.Client.EnsureField(ctx, cmd.Index, cmd.Field) if err != nil { return fmt.Errorf("Error Creating Field: %s", err) } diff --git a/field.go b/field.go index fda0ee364..93c3d5827 100644 --- a/field.go +++ b/field.go @@ -15,6 +15,7 @@ package pilosa import ( + "encoding/json" "fmt" "io/ioutil" "os" @@ -33,10 +34,10 @@ import ( const ( DefaultFieldType = FieldTypeSet - defaultCacheType = CacheTypeRanked + DefaultCacheType = CacheTypeRanked // Default ranked field cache - defaultCacheSize = 50000 + DefaultCacheSize = 50000 ) // Field types. @@ -69,19 +70,46 @@ type Field struct { Logger Logger } -// FieldOption is a functional option type for pilosa.Fielde. -type FieldOption func(f *Field) error +// FieldOption is a functional option type for pilosa.FieldOptions. +type FieldOption func(fo *FieldOptions) error -// TODO: break these out into separate Options (not a FieldOptions object) -func OptFieldFieldOptions(o FieldOptions) FieldOption { - return func(f *Field) error { - f.options = o +func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption { + return func(fo *FieldOptions) error { + if fo.Type != "" { + return errors.Errorf("field type is already set to: %s", fo.Type) + } + fo.Type = FieldTypeSet + fo.CacheType = cacheType + fo.CacheSize = cacheSize + return nil + } +} + +func OptFieldTypeInt(min, max int64) FieldOption { + return func(fo *FieldOptions) error { + if fo.Type != "" { + return errors.Errorf("field type is already set to: %s", fo.Type) + } + fo.Type = FieldTypeInt + fo.Min = min + fo.Max = max + return nil + } +} + +func OptFieldTypeTime(timeQuantum TimeQuantum) FieldOption { + return func(fo *FieldOptions) error { + if fo.Type != "" { + return errors.Errorf("field type is already set to: %s", fo.Type) + } + fo.Type = FieldTypeTime + fo.TimeQuantum = timeQuantum return nil } } // NewField returns a new instance of field. -func NewField(path, index, name string, opts ...FieldOption) (*Field, error) { +func NewField(path, index, name string, options FieldOptions) (*Field, error) { err := validateName(name) if err != nil { return nil, err @@ -99,22 +127,10 @@ func NewField(path, index, name string, opts ...FieldOption) (*Field, error) { broadcaster: NopBroadcaster, Stats: NopStatsClient, - options: FieldOptions{ - Type: DefaultFieldType, - CacheType: defaultCacheType, - CacheSize: defaultCacheSize, - }, + options: applyDefaultOptions(options), Logger: NopLogger, } - - for _, opt := range opts { - err := opt(f) - if err != nil { - return nil, errors.Wrap(err, "applying option") - } - } - return f, nil } @@ -1046,6 +1062,19 @@ type FieldOptions struct { Keys bool `json:"keys,omitempty"` } +// applyDefaultOptions returns a new FieldOptions object +// with default values if o does not contain a valid type. +func applyDefaultOptions(o FieldOptions) FieldOptions { + if o.Type == "" { + return FieldOptions{ + Type: DefaultFieldType, + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, + } + } + return o +} + // Validate ensures that FieldOption values are valid. func (o *FieldOptions) Validate() error { switch o.Type { @@ -1100,6 +1129,40 @@ func decodeFieldOptions(options *internal.FieldOptions) *FieldOptions { } } +func (o *FieldOptions) MarshalJSON() ([]byte, error) { + switch o.Type { + case FieldTypeSet: + return json.Marshal(struct { + Type string `json:"type"` + CacheType string `json:"cacheType"` + CacheSize uint32 `json:"cacheSize"` + }{ + o.Type, + o.CacheType, + o.CacheSize, + }) + case FieldTypeInt: + return json.Marshal(struct { + Type string `json:"type"` + Min int64 `json:"min"` + Max int64 `json:"max"` + }{ + o.Type, + o.Min, + o.Max, + }) + case FieldTypeTime: + return json.Marshal(struct { + Type string `json:"type"` + TimeQuantum TimeQuantum `json:"timeQuantum"` + }{ + o.Type, + o.TimeQuantum, + }) + } + return nil, errors.New("invalid field type") +} + // List of bsiGroup types. const ( bsiGroupTypeInt = "int" diff --git a/field_test.go b/field_test.go index 10e4d9a97..e3719a35a 100644 --- a/field_test.go +++ b/field_test.go @@ -24,7 +24,7 @@ import ( // Ensure field can open and retrieve a view. func TestField_CreateViewIfNotExists(t *testing.T) { - f := test.MustOpenField() + f := test.MustOpenField(pilosa.FieldOptions{}) defer f.Close() // Create view. @@ -50,10 +50,7 @@ func TestField_CreateViewIfNotExists(t *testing.T) { // Ensure field can set its time quantum. func TestField_SetTimeQuantum(t *testing.T) { - fo := pilosa.FieldOptions{ - Type: "time", - } - f := test.MustOpenField(pilosa.OptFieldFieldOptions(fo)) + f := test.MustOpenField(pilosa.FieldOptions{Type: pilosa.FieldTypeTime}) defer f.Close() // Set & retrieve time quantum. @@ -208,7 +205,7 @@ func TestField_NameRestriction(t *testing.T) { if err != nil { panic(err) } - field, err := pilosa.NewField(path, "i", ".meta") + field, err := pilosa.NewField(path, "i", ".meta", pilosa.FieldOptions{}) if field != nil { t.Fatalf("unexpected field name %s", err) } @@ -240,13 +237,13 @@ func TestField_NameValidation(t *testing.T) { panic(err) } for _, name := range validFieldNames { - _, err := pilosa.NewField(path, "i", name) + _, err := pilosa.NewField(path, "i", name, pilosa.FieldOptions{}) if err != nil { t.Fatalf("unexpected field name: %s %s", name, err) } } for _, name := range invalidFieldNames { - _, err := pilosa.NewField(path, "i", name) + _, err := pilosa.NewField(path, "i", name, pilosa.FieldOptions{}) if err == nil { t.Fatalf("expected error on field name: %s", name) } @@ -255,7 +252,7 @@ func TestField_NameValidation(t *testing.T) { // Ensure field can open and retrieve a view. func TestField_DeleteView(t *testing.T) { - f := test.MustOpenField() + f := test.MustOpenField(pilosa.FieldOptions{}) defer f.Close() viewName := pilosa.ViewStandard + "_v" diff --git a/fragment.go b/fragment.go index 28978d337..4fc24d71e 100644 --- a/fragment.go +++ b/fragment.go @@ -117,8 +117,8 @@ func NewFragment(path, index, field, view string, slice uint64) *Fragment { field: field, view: view, slice: slice, - CacheType: defaultCacheType, - CacheSize: defaultCacheSize, + CacheType: DefaultCacheType, + CacheSize: DefaultCacheSize, Logger: NopLogger, MaxOpN: defaultFragmentMaxOpN, diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 6c733ba4c..e7a77dafb 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1245,7 +1245,7 @@ func mustOpenFragment(index, field, view string, slice uint64, cacheType string) file.Close() if cacheType == "" { - cacheType = defaultCacheType + cacheType = DefaultCacheType } f := NewFragment(file.Name(), index, field, view, slice) diff --git a/http/client.go b/http/client.go index 5090072a3..378413c2f 100644 --- a/http/client.go +++ b/http/client.go @@ -334,8 +334,8 @@ func (c *InternalClient) EnsureIndex(ctx context.Context, name string, options p return err } -func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string, options pilosa.FieldOptions) error { - err := c.CreateField(ctx, indexName, fieldName, options) +func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error { + err := c.CreateField(ctx, indexName, fieldName) if err == nil || err == pilosa.ErrFieldExists { return nil } @@ -620,14 +620,15 @@ func (c *InternalClient) backupSliceNode(ctx context.Context, index, field strin } // CreateField creates a new field on the server. -func (c *InternalClient) CreateField(ctx context.Context, index, field string, opt pilosa.FieldOptions) error { +func (c *InternalClient) CreateField(ctx context.Context, index, field string) error { if index == "" { return pilosa.ErrIndexRequired } + // TODO: remove buf completely? (depends on whether importer needs to create specific field types) // Encode query request. buf, err := json.Marshal(&postFieldRequest{ - Options: opt, + //Options: opt, }) if err != nil { return errors.Wrap(err, "marshaling") diff --git a/http/handler.go b/http/handler.go index a079ae479..6580ce8f0 100644 --- a/http/handler.go +++ b/http/handler.go @@ -457,6 +457,17 @@ func (p *postIndexRequest) UnmarshalJSON(b []byte) error { return nil } +func getValidOptions(option interface{}) []string { + validOptions := []string{} + val := reflect.ValueOf(option) + for i := 0; i < val.Type().NumField(); i++ { + jsonTag := val.Type().Field(i).Tag.Get("json") + s := strings.Split(jsonTag, ",") + validOptions = append(validOptions, s[0]) + } + return validOptions +} + // Raise errors for any unknown key func validateOptions(data map[string]interface{}, validIndexOptions []string) error { for k, v := range data { @@ -597,7 +608,9 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { // Decode request. var req postFieldRequest - err := json.NewDecoder(r.Body).Decode(&req) + dec := json.NewDecoder(r.Body) + dec.DisallowUnknownFields() + err := dec.Decode(&req) if err == io.EOF { // If no data was provided (EOF), we still create the field // with default values. @@ -605,7 +618,25 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } - _, err = h.API.CreateField(r.Context(), indexName, fieldName, req.Options) + + // Validate field options. + if err := req.Options.validate(); err != nil { + http.Error(w, err.Error(), http.StatusNotAcceptable) + return + } + + // Convert json options into functional options. + var fos []pilosa.FieldOption + switch req.Options.Type { + case pilosa.FieldTypeSet: + fos = append(fos, pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize)) + case pilosa.FieldTypeInt: + fos = append(fos, pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max)) + case pilosa.FieldTypeTime: + fos = append(fos, pilosa.OptFieldTypeTime(*req.Options.TimeQuantum)) + } + + _, err = h.API.CreateField(r.Context(), indexName, fieldName, fos...) if err != nil { switch errors.Cause(err) { case pilosa.ErrIndexNotFound: @@ -623,51 +654,80 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { } } -type _postFieldRequest postFieldRequest - -// Custom Unmarshal JSON to validate request body when creating a new field. If there's new FieldOptions, -// adding it to validFieldOptions to make sure the new option is validated, otherwise the request will be failed -func (p *postFieldRequest) UnmarshalJSON(b []byte) error { - // m is an overflow map used to capture additional, unexpected keys. - m := make(map[string]interface{}) - if err := json.Unmarshal(b, &m); err != nil { - return errors.Wrap(err, "unmarshaling unexpected keys") - } - - validFieldOptions := getValidOptions(pilosa.FieldOptions{}) - err := validateOptions(m, validFieldOptions) - if err != nil { - return err - } - - // Unmarshal expected values. - var _p _postFieldRequest - if err := json.Unmarshal(b, &_p); err != nil { - return errors.Wrap(err, "unmarshalling expected keys") - } - - p.Options = _p.Options - return nil - -} - -func getValidOptions(option interface{}) []string { - validOptions := []string{} - val := reflect.ValueOf(option) - for i := 0; i < val.Type().NumField(); i++ { - jsonTag := val.Type().Field(i).Tag.Get("json") - s := strings.Split(jsonTag, ",") - validOptions = append(validOptions, s[0]) - } - return validOptions -} - type postFieldRequest struct { - Options pilosa.FieldOptions `json:"options"` + Options fieldOptions `json:"options"` } type postFieldResponse struct{} +// fieldOptions tracks pilosa.FieldOptions. It is made up of pointers to values, +// and used for input validation. +type fieldOptions struct { + Type string `json:"type,omitempty"` + CacheType *string `json:"cacheType,omitempty"` + CacheSize *uint32 `json:"cacheSize,omitempty"` + Min *int64 `json:"min,omitempty"` + Max *int64 `json:"max,omitempty"` + TimeQuantum *pilosa.TimeQuantum `json:"timeQuantum,omitempty"` + Keys *bool `json:"keys,omitempty"` +} + +func (o *fieldOptions) validate() error { + // Pointers to default values. + defaultCacheType := pilosa.DefaultCacheType + defaultCacheSize := uint32(pilosa.DefaultCacheSize) + + switch o.Type { + case pilosa.FieldTypeSet, "": + // Because FieldTypeSet is the default, its arguments are + // not required. Instead, the defaults are applied whenever + // a value does not exist. + if o.Type == "" { + o.Type = pilosa.FieldTypeSet + } + if o.CacheType == nil { + o.CacheType = &defaultCacheType + } + if o.CacheSize == nil { + o.CacheSize = &defaultCacheSize + } + if o.Min != nil { + return errors.New("min does not apply to field type set") + } else if o.Max != nil { + return errors.New("max does not apply to field type set") + } else if o.TimeQuantum != nil { + return errors.New("timeQuantum does not apply to field type set") + } + case pilosa.FieldTypeInt: + if o.CacheType != nil { + return errors.New("cacheType does not apply to field type int") + } else if o.CacheSize != nil { + return errors.New("cacheSize does not apply to field type int") + } else if o.Min == nil { + return errors.New("min is required for field type int") + } else if o.Max == nil { + return errors.New("max is required for field type int") + } else if o.TimeQuantum != nil { + return errors.New("timeQuantum does not apply to field type int") + } + case pilosa.FieldTypeTime: + if o.CacheType != nil { + return errors.New("cacheType does not apply to field type time") + } else if o.CacheSize != nil { + return errors.New("cacheSize does not apply to field type time") + } else if o.Min != nil { + return errors.New("min does not apply to field type time") + } else if o.Max != nil { + return errors.New("max does not apply to field type time") + } else if o.TimeQuantum == nil { + return errors.New("timeQuantum is required for field type time") + } + default: + return errors.Errorf("invalid field type: %s", o.Type) + } + return nil +} + // handleDeleteField handles DELETE /field request. func (h *Handler) handleDeleteField(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { diff --git a/http/handler_internal_test.go b/http/handler_internal_test.go index fa7c23060..f7f95aeb0 100644 --- a/http/handler_internal_test.go +++ b/http/handler_internal_test.go @@ -15,6 +15,7 @@ package http import ( + "bytes" "encoding/json" "reflect" "testing" @@ -59,31 +60,121 @@ func TestPostIndexRequestUnmarshalJSON(t *testing.T) { // Test custom UnmarshalJSON for postFieldRequest object func TestPostFieldRequestUnmarshalJSON(t *testing.T) { + foo := "foo" tests := []struct { json string expected postFieldRequest err string }{ - {json: `{"options": {}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{}}}, - {json: `{"options": 4}`, err: "options is not map[string]interface{}"}, - {json: `{"option": {}}`, err: "Unknown key: option:map[]"}, - {json: `{"options": {"badKey": "test"}}`, err: "Unknown key: badKey:test"}, - {json: `{"options": {"inverseEnabled": true}}`, err: "Unknown key: inverseEnabled:true"}, - {json: `{"options": {"cacheType": "type"}}`, expected: postFieldRequest{Options: pilosa.FieldOptions{CacheType: "type"}}}, - {json: `{"options": {"inverse": true, "cacheType": "type"}}`, err: "Unknown key: inverse:true"}, + {json: `{"options": {}}`, expected: postFieldRequest{}}, + {json: `{"options": 4}`, err: "json: cannot unmarshal number into Go struct field postFieldRequest.options of type http.fieldOptions"}, + {json: `{"option": {}}`, err: `json: unknown field "option"`}, + {json: `{"options": {"badKey": "test"}}`, err: `json: unknown field "badKey"`}, + {json: `{"options": {"inverseEnabled": true}}`, err: `json: unknown field "inverseEnabled"`}, + {json: `{"options": {"cacheType": "foo"}}`, expected: postFieldRequest{Options: fieldOptions{CacheType: &foo}}}, + {json: `{"options": {"inverse": true, "cacheType": "foo"}}`, err: `json: unknown field "inverse"`}, } - for _, test := range tests { + for i, test := range tests { actual := &postFieldRequest{} - err := json.Unmarshal([]byte(test.json), actual) + dec := json.NewDecoder(bytes.NewReader([]byte(test.json))) + dec.DisallowUnknownFields() + err := dec.Decode(actual) if err != nil { if test.err == "" || test.err != err.Error() { - t.Errorf("expected error: %v, but got result: %v", test.err, err) + t.Errorf("test %d: expected error: %v, but got result: %v", i, test.err, err) } } if test.err == "" { if !reflect.DeepEqual(*actual, test.expected) { - t.Errorf("expected: %v, but got: %v", test.expected, *actual) + t.Errorf("test %d: expected: %v, but got: %v", i, test.expected, *actual) + } + } + + } +} + +func stringPtr(s string) *string { + return &s +} + +func int64Ptr(i int64) *int64 { + return &i +} + +// Test fieldOption validation. +func TestFieldOptionValidation(t *testing.T) { + //foo := "foo" + //set := "set" + timeQuantum := pilosa.TimeQuantum("YMD") + defaultCacheSize := uint32(pilosa.DefaultCacheSize) + tests := []struct { + json string + expected postFieldRequest + err string + }{ + // FieldType: Set + {json: `{"options": {}}`, expected: postFieldRequest{Options: fieldOptions{ + Type: pilosa.FieldTypeSet, + CacheType: stringPtr(pilosa.DefaultCacheType), + CacheSize: &defaultCacheSize, + }}}, + {json: `{"options": {"type": "set"}}`, expected: postFieldRequest{Options: fieldOptions{ + Type: pilosa.FieldTypeSet, + CacheType: stringPtr(pilosa.DefaultCacheType), + CacheSize: &defaultCacheSize, + }}}, + {json: `{"options": {"type": "set", "cacheType": "lru"}}`, expected: postFieldRequest{Options: fieldOptions{ + Type: pilosa.FieldTypeSet, + CacheType: stringPtr("lru"), + CacheSize: &defaultCacheSize, + }}}, + {json: `{"options": {"type": "set", "min": 0}}`, err: "min does not apply to field type set"}, + {json: `{"options": {"type": "set", "max": 100}}`, err: "max does not apply to field type set"}, + {json: `{"options": {"type": "set", "timeQuantum": "YMD"}}`, err: "timeQuantum does not apply to field type set"}, + + // FieldType: Int + {json: `{"options": {"type": "int"}}`, err: "min is required for field type int"}, + {json: `{"options": {"type": "int", "min": 0}}`, err: "max is required for field type int"}, + {json: `{"options": {"type": "int", "min": 0, "max": 1000}}`, expected: postFieldRequest{Options: fieldOptions{ + Type: pilosa.FieldTypeInt, + Min: int64Ptr(0), + Max: int64Ptr(1000), + }}}, + {json: `{"options": {"type": "int", "min": 0, "max": 1000, "cacheType": "ranked"}}`, err: "cacheType does not apply to field type int"}, + {json: `{"options": {"type": "int", "min": 0, "max": 1000, "cacheSize": 1000}}`, err: "cacheSize does not apply to field type int"}, + {json: `{"options": {"type": "int", "min": 0, "max": 1000, "timeQuantum": "YMD"}}`, err: "timeQuantum does not apply to field type int"}, + + // FieldType: Time + {json: `{"options": {"type": "time"}}`, err: "timeQuantum is required for field type time"}, + {json: `{"options": {"type": "time", "timeQuantum": "YMD"}}`, expected: postFieldRequest{Options: fieldOptions{ + Type: pilosa.FieldTypeTime, + TimeQuantum: &timeQuantum, + }}}, + {json: `{"options": {"type": "time", "timeQuantum": "YMD", "min": 0}}`, err: "min does not apply to field type time"}, + {json: `{"options": {"type": "time", "timeQuantum": "YMD", "max": 1000}}`, err: "max does not apply to field type time"}, + {json: `{"options": {"type": "time", "timeQuantum": "YMD", "cacheType": "ranked"}}`, err: "cacheType does not apply to field type time"}, + {json: `{"options": {"type": "time", "timeQuantum": "YMD", "cacheSize": 1000}}`, err: "cacheSize does not apply to field type time"}, + } + for i, test := range tests { + actual := &postFieldRequest{} + dec := json.NewDecoder(bytes.NewReader([]byte(test.json))) + dec.DisallowUnknownFields() + err := dec.Decode(actual) + if err != nil { + t.Errorf("test %d: %v", i, err) + } + + // Validate field options. + if err := actual.Options.validate(); err != nil { + if test.err == "" || test.err != err.Error() { + t.Errorf("test %d: expected error: %v, but got result: %v", i, test.err, err) + } + } + + if test.err == "" { + if !reflect.DeepEqual(*actual, test.expected) { + t.Errorf("test %d: expected: %v, but got: %v", i, test.expected, *actual) } } diff --git a/index.go b/index.go index 68c48c829..10e954aef 100644 --- a/index.go +++ b/index.go @@ -335,7 +335,7 @@ func (i *Index) createField(name string, opt FieldOptions) (*Field, error) { } func (i *Index) newField(path, name string) (*Field, error) { - f, err := NewField(path, i.name, name) + f, err := NewField(path, i.name, name, FieldOptions{}) // TODO: NewField should be un-exported along with FieldOptions if err != nil { return nil, err } diff --git a/server/cluster_test.go b/server/cluster_test.go index 73de82279..b9a1917a9 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -54,7 +54,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Create indexes and fields on one node. if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil { + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal(err) } @@ -209,7 +209,7 @@ func TestClusterResize_AddNode(t *testing.T) { // Create indexes and fields on one node. if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil { + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal(err) } @@ -253,7 +253,7 @@ func TestClusterResize_AddNode(t *testing.T) { // Create indexes and fields on one node. if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil { + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal(err) } @@ -305,7 +305,7 @@ func TestClusterResize_AddNode(t *testing.T) { // Create indexes and fields on one node. if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil { + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal(err) } @@ -458,7 +458,7 @@ func TestClusterResize_RemoveNode(t *testing.T) { // Create indexes and fields on one node. if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{}); err != nil { + } else if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal(err) } diff --git a/server/server_test.go b/server/server_test.go index 58883286f..b3067db87 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -54,7 +54,7 @@ func TestMain_Set_Quick(t *testing.T) { if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) } - if err := client.CreateField(context.Background(), "i", cmd.Field, pilosa.FieldOptions{}); err != nil && err != pilosa.ErrFieldExists { + if err := client.CreateField(context.Background(), "i", cmd.Field); err != nil && err != pilosa.ErrFieldExists { t.Fatal(err) } if _, err := m.Query("i", "", fmt.Sprintf(`Set(%d, %s=%d)`, cmd.ColumnID, cmd.Field, cmd.ID)); err != nil { @@ -123,11 +123,11 @@ func TestMain_SetRowAttrs(t *testing.T) { client := m.Client() if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client.CreateField(context.Background(), "i", "x", pilosa.FieldOptions{}); err != nil { + } else if err := client.CreateField(context.Background(), "i", "x"); err != nil { t.Fatal(err) - } else if err := client.CreateField(context.Background(), "i", "z", pilosa.FieldOptions{}); err != nil { + } else if err := client.CreateField(context.Background(), "i", "z"); err != nil { t.Fatal(err) - } else if err := client.CreateField(context.Background(), "i", "neg", pilosa.FieldOptions{}); err != nil { + } else if err := client.CreateField(context.Background(), "i", "neg"); err != nil { t.Fatal(err) } @@ -200,7 +200,7 @@ func TestMain_SetColumnAttrs(t *testing.T) { client := m.Client() if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal(err) - } else if err := client.CreateField(context.Background(), "i", "x", pilosa.FieldOptions{}); err != nil { + } else if err := client.CreateField(context.Background(), "i", "x"); err != nil { t.Fatal(err) } @@ -271,9 +271,10 @@ func TestMain_RecalculateHashes(t *testing.T) { if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { t.Fatal("create index:", err) } - if err := client0.CreateField(context.Background(), "i", "f", pilosa.FieldOptions{CacheType: "ranked"}); err != nil { + if err := client0.CreateField(context.Background(), "i", "f"); err != nil { t.Fatal("create field:", err) } + return // Set some columns data := []string{} diff --git a/server_test.go b/server_test.go index 402d4de7d..9f076369b 100644 --- a/server_test.go +++ b/server_test.go @@ -33,7 +33,8 @@ func TestMonitorAntiEntropy(t *testing.T) { if err != nil { t.Fatalf("creating index: %v", err) } - err = client.CreateField(context.Background(), "balh", "fralh", pilosa.FieldOptions{}) + + err = client.CreateField(context.Background(), "balh", "fralh") if err != nil { t.Fatalf("creating field: %v", err) } diff --git a/test/field.go b/test/field.go index 9a83be2de..345deadf8 100644 --- a/test/field.go +++ b/test/field.go @@ -28,12 +28,12 @@ type Field struct { } // NewField returns a new instance of Field d/0. -func NewField(opt ...pilosa.FieldOption) *Field { +func NewField(options pilosa.FieldOptions) *Field { path, err := ioutil.TempDir("", "pilosa-field-") if err != nil { panic(err) } - field, err := pilosa.NewField(path, "i", "f", opt...) + field, err := pilosa.NewField(path, "i", "f", options) if err != nil { panic(err) } @@ -41,8 +41,8 @@ func NewField(opt ...pilosa.FieldOption) *Field { } // MustOpenField returns a new, opened field at a temporary path. Panic on error. -func MustOpenField(opt ...pilosa.FieldOption) *Field { - f := NewField(opt...) +func MustOpenField(options pilosa.FieldOptions) *Field { + f := NewField(options) if err := f.Open(); err != nil { panic(err) } @@ -63,7 +63,7 @@ func (f *Field) Reopen() error { } path, index, name := f.Path(), f.Index(), f.Name() - f.Field, err = pilosa.NewField(path, index, name) + f.Field, err = pilosa.NewField(path, index, name, pilosa.FieldOptions{}) if err != nil { return err } @@ -76,7 +76,7 @@ func (f *Field) Reopen() error { // Ensure field can set its cache func TestField_SetCacheSize(t *testing.T) { - f := MustOpenField() + f := MustOpenField(pilosa.FieldOptions{}) defer f.Close() cacheSize := uint32(100) diff --git a/view.go b/view.go index 428fc6f54..732ecd485 100644 --- a/view.go +++ b/view.go @@ -73,7 +73,7 @@ func NewView(path, index, field, name string, cacheSize uint32) *View { name: name, cacheSize: cacheSize, - cacheType: defaultCacheType, + cacheType: DefaultCacheType, fragments: make(map[uint64]*Fragment), broadcaster: NopBroadcaster, diff --git a/view_internal_test.go b/view_internal_test.go index 48df030db..d0e8bfdd1 100644 --- a/view_internal_test.go +++ b/view_internal_test.go @@ -26,7 +26,7 @@ func mustOpenView(index, field, name string) *View { panic(err) } - v := NewView(path, index, field, name, defaultCacheSize) + v := NewView(path, index, field, name, DefaultCacheSize) if err := v.open(); err != nil { panic(err) } From 028e95d914942fbb25237f575c46e74d0ca0a517 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Tue, 26 Jun 2018 10:37:56 -0500 Subject: [PATCH 2/6] Allow a single functional option for field options. Move field type specific validation to functional options. --- api.go | 13 +++++++------ ctl/import.go | 1 - field.go | 25 ++++++------------------- http/handler.go | 10 +++++----- http/handler_internal_test.go | 2 -- index.go | 5 ----- test/holder.go | 2 +- 7 files changed, 19 insertions(+), 39 deletions(-) diff --git a/api.go b/api.go index dd0d5fca4..66add8863 100644 --- a/api.go +++ b/api.go @@ -244,17 +244,18 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error { } // CreateField makes the named field in the named index with the given options. -func (api *API) CreateField(ctx context.Context, indexName string, fieldName string, opts ...FieldOption) (*Field, error) { +// 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(ctx context.Context, indexName string, fieldName string, opts FieldOption) (*Field, error) { if err := api.validate(apiCreateField); err != nil { return nil, errors.Wrap(err, "validating api method") } + // Apply functional option. fo := FieldOptions{} - for _, opt := range opts { - err := opt(&fo) - if err != nil { - return nil, errors.Wrap(err, "applying option") - } + err := opts(&fo) + if err != nil { + return nil, errors.Wrap(err, "applying option") } // Find index. diff --git a/ctl/import.go b/ctl/import.go index ad4befbfd..05659fa27 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -42,7 +42,6 @@ type ImportCommand struct { // Options for index & field to be created if they don't exist IndexOptions pilosa.IndexOptions - //FieldOptions pilosa.FieldOptions // CreateSchema ensures the schema exists before import CreateSchema bool diff --git a/field.go b/field.go index 93c3d5827..4fd684682 100644 --- a/field.go +++ b/field.go @@ -90,6 +90,9 @@ func OptFieldTypeInt(min, max int64) FieldOption { if fo.Type != "" { return errors.Errorf("field type is already set to: %s", fo.Type) } + if min > max { + return ErrInvalidBSIGroupRange + } fo.Type = FieldTypeInt fo.Min = min fo.Max = max @@ -102,6 +105,9 @@ func OptFieldTypeTime(timeQuantum TimeQuantum) FieldOption { if fo.Type != "" { return errors.Errorf("field type is already set to: %s", fo.Type) } + if !timeQuantum.Valid() { + return ErrInvalidTimeQuantum + } fo.Type = FieldTypeTime fo.TimeQuantum = timeQuantum return nil @@ -1075,25 +1081,6 @@ func applyDefaultOptions(o FieldOptions) FieldOptions { return o } -// Validate ensures that FieldOption values are valid. -func (o *FieldOptions) Validate() error { - switch o.Type { - case FieldTypeSet, "": - // TODO: cacheType, cacheSize validation - case FieldTypeInt: - if o.Min > o.Max { - return ErrInvalidBSIGroupRange - } - case FieldTypeTime: - if o.TimeQuantum == "" || !o.TimeQuantum.Valid() { - return ErrInvalidTimeQuantum - } - default: - return errors.New("invalid field type") - } - return nil -} - // Encode converts o into its internal representation. func (o *FieldOptions) Encode() *internal.FieldOptions { return encodeFieldOptions(o) diff --git a/http/handler.go b/http/handler.go index 6580ce8f0..bbe784f69 100644 --- a/http/handler.go +++ b/http/handler.go @@ -626,17 +626,17 @@ func (h *Handler) handlePostField(w http.ResponseWriter, r *http.Request) { } // Convert json options into functional options. - var fos []pilosa.FieldOption + var fos pilosa.FieldOption switch req.Options.Type { case pilosa.FieldTypeSet: - fos = append(fos, pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize)) + fos = pilosa.OptFieldTypeSet(*req.Options.CacheType, *req.Options.CacheSize) case pilosa.FieldTypeInt: - fos = append(fos, pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max)) + fos = pilosa.OptFieldTypeInt(*req.Options.Min, *req.Options.Max) case pilosa.FieldTypeTime: - fos = append(fos, pilosa.OptFieldTypeTime(*req.Options.TimeQuantum)) + fos = pilosa.OptFieldTypeTime(*req.Options.TimeQuantum) } - _, err = h.API.CreateField(r.Context(), indexName, fieldName, fos...) + _, err = h.API.CreateField(r.Context(), indexName, fieldName, fos) if err != nil { switch errors.Cause(err) { case pilosa.ErrIndexNotFound: diff --git a/http/handler_internal_test.go b/http/handler_internal_test.go index f7f95aeb0..071ac1926 100644 --- a/http/handler_internal_test.go +++ b/http/handler_internal_test.go @@ -104,8 +104,6 @@ func int64Ptr(i int64) *int64 { // Test fieldOption validation. func TestFieldOptionValidation(t *testing.T) { - //foo := "foo" - //set := "set" timeQuantum := pilosa.TimeQuantum("YMD") defaultCacheSize := uint32(pilosa.DefaultCacheSize) tests := []struct { diff --git a/index.go b/index.go index 10e954aef..a118b1808 100644 --- a/index.go +++ b/index.go @@ -301,11 +301,6 @@ func (i *Index) createField(name string, opt FieldOptions) (*Field, error) { return nil, ErrInvalidCacheType } - // Validate options. - if err := opt.Validate(); err != nil { - return nil, errors.Wrap(err, "validating options") - } - // Initialize field. f, err := i.newField(i.FieldPath(name), name) if err != nil { diff --git a/test/holder.go b/test/holder.go index 7bae8afaa..648d910ed 100644 --- a/test/holder.go +++ b/test/holder.go @@ -92,7 +92,7 @@ func (h *Holder) MustCreateFieldIfNotExists(index, field string) *Field { // MustCreateRankedFragmentIfNotExists returns a given fragment with a ranked cache. Panic on error. func (h *Holder) MustCreateRankedFragmentIfNotExists(index, field, view string, slice uint64) *Fragment { idx := h.MustCreateIndexIfNotExists(index, pilosa.IndexOptions{}) - f, err := idx.CreateFieldIfNotExists(field, pilosa.FieldOptions{CacheType: pilosa.CacheTypeRanked}) + f, err := idx.CreateFieldIfNotExists(field, pilosa.FieldOptions{}) if err != nil { panic(err) } From 5d43d414f7f14185e029bc2ea1adf0762a7a558d Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Wed, 27 Jun 2018 16:59:35 -0500 Subject: [PATCH 3/6] Fix a few data races --- http/translator_test.go | 17 +++++++++++------ translate.go | 4 +++- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/http/translator_test.go b/http/translator_test.go index d317944e4..fdfc93f3b 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -5,6 +5,7 @@ import ( "io" "io/ioutil" gohttp "net/http" + "sync/atomic" "testing" "time" @@ -37,9 +38,10 @@ func TestTranslateStore_Reader(t *testing.T) { return 0, nil } } - var closeInvoked bool + closeInvoked := atomic.Value{} + closeInvoked.Store(false) mrc.CloseFunc = func() error { - closeInvoked = true + closeInvoked.Store(true) return nil } @@ -85,7 +87,7 @@ func TestTranslateStore_Reader(t *testing.T) { t.Fatal(err) } - if !closeInvoked { + if !closeInvoked.Load().(bool) { t.Fatal("expected server close") } }) @@ -100,9 +102,12 @@ func TestTranslateStore_Reader(t *testing.T) { <-done return 0, io.EOF } - var closeInvoked bool + + closeInvoked := atomic.Value{} + closeInvoked.Store(false) + mrc.CloseFunc = func() error { - closeInvoked = true + closeInvoked.Store(true) return nil } @@ -127,7 +132,7 @@ func TestTranslateStore_Reader(t *testing.T) { // Cancel the context and check if server is closed. cancel() time.Sleep(100 * time.Millisecond) - if !closeInvoked { + if !closeInvoked.Load().(bool) { t.Fatal("expected server-side close") } }) diff --git a/translate.go b/translate.go index 398055fdb..7c716640b 100644 --- a/translate.go +++ b/translate.go @@ -306,11 +306,13 @@ func (s *TranslateFile) replicate(ctx context.Context) error { } else if err != nil { return err } - + s.mu.Lock() // Write to local store. if err := s.appendEntry(&entry); err != nil { + s.mu.Unlock() return err } + s.mu.Unlock() } } From 25be5c0f2fe0ca18605d3ac1dd587773c5d57e48 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Thu, 28 Jun 2018 10:38:37 -0500 Subject: [PATCH 4/6] Use channel to notify on server close instead of atomic.Value. Ensure CloseFunc() only called once. --- http/translator_test.go | 46 +++++++++++++++++++++++------------------ mock/mock.go | 9 +++++++- 2 files changed, 34 insertions(+), 21 deletions(-) diff --git a/http/translator_test.go b/http/translator_test.go index fdfc93f3b..1eea725a2 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -5,7 +5,6 @@ import ( "io" "io/ioutil" gohttp "net/http" - "sync/atomic" "testing" "time" @@ -16,6 +15,17 @@ import ( "github.com/pilosa/pilosa/test" ) +func newMockReadCloser() *mock.ReadCloser { + return &mock.ReadCloser{ + ReadFunc: func(p []byte) (int, error) { + return 0, io.EOF + }, + CloseFunc: func() error { + return nil + }, + } +} + func TestTranslateStore_Reader(t *testing.T) { // Ensure client can connect and stream the translate store data. t.Run("OK", func(t *testing.T) { @@ -38,10 +48,9 @@ func TestTranslateStore_Reader(t *testing.T) { return 0, nil } } - closeInvoked := atomic.Value{} - closeInvoked.Store(false) + closeInvoked := make(chan struct{}) mrc.CloseFunc = func() error { - closeInvoked.Store(true) + close(closeInvoked) return nil } @@ -57,15 +66,7 @@ func TestTranslateStore_Reader(t *testing.T) { } return &mrc, nil } - mrc2 := mock.ReadCloser{ - ReadFunc: func(p []byte) (int, error) { - return 0, io.EOF - }, - CloseFunc: func() error { - return nil - }, - } - return &mrc2, nil + return newMockReadCloser(), nil } opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore)) @@ -87,8 +88,11 @@ func TestTranslateStore_Reader(t *testing.T) { t.Fatal(err) } - if !closeInvoked.Load().(bool) { + select { + case <-time.NewTimer(time.Millisecond * 100).C: t.Fatal("expected server close") + case <-closeInvoked: + return } }) @@ -103,15 +107,15 @@ func TestTranslateStore_Reader(t *testing.T) { return 0, io.EOF } - closeInvoked := atomic.Value{} - closeInvoked.Store(false) + closeInvoked := make(chan struct{}) mrc.CloseFunc = func() error { - closeInvoked.Store(true) + close(closeInvoked) return nil } var translateStore mock.TranslateStore + translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) { return &mrc, nil } @@ -131,9 +135,11 @@ func TestTranslateStore_Reader(t *testing.T) { // Cancel the context and check if server is closed. cancel() - time.Sleep(100 * time.Millisecond) - if !closeInvoked.Load().(bool) { - t.Fatal("expected server-side close") + select { + case <-time.NewTimer(time.Millisecond * 100).C: + t.Fatal("expected server close") + case <-closeInvoked: + return } }) }) diff --git a/mock/mock.go b/mock/mock.go index 46469c2f9..97ebf8641 100644 --- a/mock/mock.go +++ b/mock/mock.go @@ -1,8 +1,11 @@ package mock +import "sync" + type ReadCloser struct { ReadFunc func(p []byte) (int, error) CloseFunc func() error + once sync.Once } func (rc *ReadCloser) Read(p []byte) (int, error) { @@ -10,5 +13,9 @@ func (rc *ReadCloser) Read(p []byte) (int, error) { } func (rc *ReadCloser) Close() error { - return rc.CloseFunc() + var err error = nil + rc.once.Do(func() { + err = rc.CloseFunc() + }) + return err } From 824474160e28b29d0919f4d3882ac99f9879a3f9 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Wed, 27 Jun 2018 17:11:55 -0500 Subject: [PATCH 5/6] Enhance test utilities (introduce Cluster type, improve naming) --- ctl/export_test.go | 2 +- ctl/import_test.go | 8 +- executor_test.go | 4 +- http/client_test.go | 6 +- http/translator_test.go | 6 +- server/cluster_test.go | 30 +++---- server/handler_test.go | 4 +- server/server_test.go | 8 +- server_test.go | 2 +- stats_test.go | 2 +- test/pilosa.go | 168 ++++++++++++++++++++++++---------------- test/pilosa_test.go | 4 +- 12 files changed, 141 insertions(+), 103 deletions(-) diff --git a/ctl/export_test.go b/ctl/export_test.go index 8960702f6..e959da6bc 100644 --- a/ctl/export_test.go +++ b/ctl/export_test.go @@ -44,7 +44,7 @@ func TestExportCommand_Validation(t *testing.T) { } func TestExportCommand_Run(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] buf := bytes.Buffer{} stdin, stdout, stderr := GetIO(buf) diff --git a/ctl/import_test.go b/ctl/import_test.go index 522de0eba..32792c02d 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -61,7 +61,7 @@ func TestImportCommand_Run(t *testing.T) { t.Fatal(err) } - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] cm.Host = cmd.Server.URI.HostPort() cm.Index = "i" @@ -86,7 +86,7 @@ func TestImportCommand_RunValue(t *testing.T) { t.Fatal(err) } - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] cm.Host = cmd.Server.URI.HostPort() http.DefaultClient.Do(MustNewHTTPRequest("POST", "http://"+cm.Host+"/index/i", strings.NewReader(""))) @@ -102,7 +102,7 @@ func TestImportCommand_RunValue(t *testing.T) { } func TestImportCommand_InvalidFile(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] buf := bytes.Buffer{} stdin, stdout, stderr := GetIO(buf) @@ -176,7 +176,7 @@ func GetIO(buf bytes.Buffer) (io.Reader, io.Writer, io.Writer) { } func TestImportCommand_BugOverwriteValue(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] buf := bytes.Buffer{} stdin, stdout, stderr := GetIO(buf) diff --git a/executor_test.go b/executor_test.go index 022219235..250d8b13d 100644 --- a/executor_test.go +++ b/executor_test.go @@ -267,7 +267,7 @@ func TestExecutor_Execute_Count(t *testing.T) { // Ensure a set query can be executed. func TestExecutor_Execute_SetBit(t *testing.T) { t.Run("ID", func(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} hldr.SetBit("i", "f", 1, 0) @@ -312,7 +312,7 @@ func TestExecutor_Execute_SetBit(t *testing.T) { }) t.Run("Keys", func(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true}) diff --git a/http/client_test.go b/http/client_test.go index 3ca4cde50..7363879ce 100644 --- a/http/client_test.go +++ b/http/client_test.go @@ -219,7 +219,7 @@ func TestClient_MultiNode(t *testing.T) { // Ensure client can bulk import data. func TestClient_Import(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] host := cmd.URL() holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} @@ -249,7 +249,7 @@ func TestClient_Import(t *testing.T) { // Ensure client can bulk import value data. func TestClient_ImportValue(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] host := cmd.URL() holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} @@ -321,7 +321,7 @@ func TestClient_ImportValue(t *testing.T) { // Ensure client can retrieve a list of all checksums for blocks in a fragment. func TestClient_FragmentBlocks(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} diff --git a/http/translator_test.go b/http/translator_test.go index 1eea725a2..a3a5e8603 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -70,7 +70,7 @@ func TestTranslateStore_Reader(t *testing.T) { } opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore)) - main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0] + main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0] defer main.Close() @@ -121,7 +121,7 @@ func TestTranslateStore_Reader(t *testing.T) { } opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore)) - main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0] + main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0] defer main.Close() defer close(done) @@ -152,7 +152,7 @@ func TestTranslateStore_Reader(t *testing.T) { } opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore)) - main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0] + main := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0] defer main.Close() _, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0) diff --git a/server/cluster_test.go b/server/cluster_test.go index 825911ace..d8c2d7802 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -31,7 +31,7 @@ import ( // Ensure program can send/receive broadcast messages. func TestMain_SendReceiveMessage(t *testing.T) { - ms := test.MustRunMainWithCluster(t, 2) + ms := test.MustRunCluster(t, 2) m0, m1 := ms[0], ms[1] defer m0.Close() defer m1.Close() @@ -116,7 +116,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Ensure that an empty node comes up in a NORMAL state. func TestClusterResize_EmptyNode(t *testing.T) { - m0 := test.MustRunMain() + m0 := test.MustRunCommand() defer m0.Close() if m0.API.State() != pilosa.ClusterStateNormal { @@ -126,7 +126,7 @@ func TestClusterResize_EmptyNode(t *testing.T) { // Ensure that a cluster of empty nodes comes up in a NORMAL state. func TestClusterResize_EmptyNodes(t *testing.T) { - clus := test.MustRunMainWithCluster(t, 2) + clus := test.MustRunCluster(t, 2) defer clus[0].Close() defer clus[1].Close() @@ -140,7 +140,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) { // Ensure that adding a node correctly resizes the cluster. func TestClusterResize_AddNode(t *testing.T) { t.Run("NoData", func(t *testing.T) { - clus := test.MustRunMainWithCluster(t, 2) + clus := test.MustRunCluster(t, 2) if !checkClusterState(clus[0], pilosa.ClusterStateNormal, 1000) { t.Fatalf("unexpected node0 cluster state: %s", clus[0].API.State()) @@ -150,7 +150,7 @@ func TestClusterResize_AddNode(t *testing.T) { }) t.Run("WithIndex", func(t *testing.T) { // Configure node0 - m0 := test.MustRunMainWithCluster(t, 1)[0] + m0 := test.MustRunCluster(t, 1)[0] defer m0.Close() seed := m0.GossipAddress() @@ -166,7 +166,7 @@ func TestClusterResize_AddNode(t *testing.T) { } // Configure node1 - m1 := test.NewMainWithCluster(false) + m1 := test.NewCommandNode(false) m1.Config.Gossip.Port = "0" m1.Config.Gossip.Seeds = []string{seed} err := m1.Start() @@ -183,7 +183,7 @@ func TestClusterResize_AddNode(t *testing.T) { }) t.Run("ContinuousSlices", func(t *testing.T) { // Configure node0 - m0 := test.MustRunMainWithCluster(t, 1)[0] + m0 := test.MustRunCluster(t, 1)[0] defer m0.Close() seed := m0.GossipAddress() @@ -207,7 +207,7 @@ func TestClusterResize_AddNode(t *testing.T) { } // Configure node1 - m1 := test.NewMainWithCluster(false) + m1 := test.NewCommandNode(false) m1.Config.Gossip.Port = "0" m1.Config.Gossip.Seeds = []string{seed} err := m1.Start() @@ -224,7 +224,7 @@ func TestClusterResize_AddNode(t *testing.T) { }) t.Run("SkippedSlice", func(t *testing.T) { // Configure node0 - m0 := test.MustRunMainWithCluster(t, 1)[0] + m0 := test.MustRunCluster(t, 1)[0] defer m0.Close() seed := m0.GossipAddress() @@ -248,7 +248,7 @@ func TestClusterResize_AddNode(t *testing.T) { } // Configure node1 - m1 := test.NewMainWithCluster(false) + m1 := test.NewCommandNode(false) m1.Config.Gossip.Port = "0" m1.Config.Gossip.Seeds = []string{seed} err := m1.Start() @@ -269,7 +269,7 @@ func TestClusterResize_AddNode(t *testing.T) { func TestCluster_GossipMembership(t *testing.T) { t.Run("Node0Down", func(t *testing.T) { // Configure node0 - m0 := test.MustRunMainWithCluster(t, 1)[0] + m0 := test.MustRunCluster(t, 1)[0] defer m0.Close() seed := m0.GossipAddress() @@ -277,7 +277,7 @@ func TestCluster_GossipMembership(t *testing.T) { var eg errgroup.Group // Configure node1 - m1 := test.NewMainWithCluster(false) + m1 := test.NewCommandNode(false) defer m1.Close() eg.Go(func() error { m1.Config.Gossip.Port = "0" @@ -291,7 +291,7 @@ func TestCluster_GossipMembership(t *testing.T) { }) // Configure node1 - m2 := test.NewMainWithCluster(false) + m2 := test.NewCommandNode(false) defer m2.Close() eg.Go(func() error { m2.Config.Gossip.Port = "0" @@ -324,7 +324,7 @@ func TestCluster_GossipMembership(t *testing.T) { } func TestClusterResize_RemoveNode(t *testing.T) { - cluster := test.MustRunMainWithCluster(t, 3) + cluster := test.MustRunCluster(t, 3) m0 := cluster[0] m1 := cluster[1] @@ -410,7 +410,7 @@ func TestClusterResize_RemoveNode(t *testing.T) { // checkClusterState polls a given cluster for its state until it // receives a matching state. It polls up to n times before returning. -func checkClusterState(m *test.Main, state string, n int) bool { +func checkClusterState(m *test.Command, state string, n int) bool { for i := 0; i < n; i++ { if m.API.State() == state { return true diff --git a/server/handler_test.go b/server/handler_test.go index cc21b8825..2981f375c 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -37,7 +37,7 @@ import ( // Ensure the handler returns "not found" for invalid paths. func TestHandler_Endpoints(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] h := cmd.Handler.(*http.Handler).Handler holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} @@ -566,7 +566,7 @@ func TestHandler_Endpoints(t *testing.T) { t.Fatalf("CORS preflight status should be 405, but is %v", result.StatusCode) } - clus := test.MustRunMainWithCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})}) + clus := test.MustRunCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})}) w = httptest.NewRecorder() h := clus[0].Handler.(*http.Handler).Handler h.ServeHTTP(w, req) diff --git a/server/server_test.go b/server/server_test.go index b3067db87..8b1e29acc 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -40,7 +40,7 @@ func TestMain_Set_Quick(t *testing.T) { } if err := quick.Check(func(cmds []SetCommand) bool { - m := test.MustRunMain() + m := test.MustRunCommand() defer m.Close() // Create client. @@ -116,7 +116,7 @@ func TestMain_Set_Quick(t *testing.T) { // Ensure program can set row attributes and retrieve them. func TestMain_SetRowAttrs(t *testing.T) { - m := test.MustRunMain() + m := test.MustRunCommand() defer m.Close() // Create fields. @@ -193,7 +193,7 @@ func TestMain_SetRowAttrs(t *testing.T) { // Ensure program can set column attributes and retrieve them. func TestMain_SetColumnAttrs(t *testing.T) { - m := test.MustRunMain() + m := test.MustRunCommand() defer m.Close() // Create fields. @@ -264,7 +264,7 @@ func tempMkdir(t *testing.T) string { func TestMain_RecalculateHashes(t *testing.T) { const clusterSize = 5 - cluster := test.MustRunMainWithCluster(t, clusterSize) + cluster := test.MustRunCluster(t, clusterSize) // Create the schema. client0 := cluster[0].Client() diff --git a/server_test.go b/server_test.go index 711572e25..04f8a6171 100644 --- a/server_test.go +++ b/server_test.go @@ -28,7 +28,7 @@ import ( // pilosa.Server was not having its remoteClient field set by an option and so // it was using a nil client in monitorAntiEntropy. func TestMonitorAntiEntropy(t *testing.T) { - cluster := test.MustRunMainWithCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)}) + cluster := test.MustRunCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)}) client := cluster[1].Client() err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{}) if err != nil { diff --git a/stats_test.go b/stats_test.go index 271057d9b..932672e3a 100644 --- a/stats_test.go +++ b/stats_test.go @@ -209,7 +209,7 @@ func TestStatsCount_SetProfileAttrs(t *testing.T) { } func TestStatsCount_APICalls(t *testing.T) { - cmd := test.MustRunMainWithCluster(t, 1)[0] + cmd := test.MustRunCluster(t, 1)[0] h := cmd.Handler.(*http.Handler).Handler holder := cmd.Server.Holder() hldr := test.Holder{Holder: holder} diff --git a/test/pilosa.go b/test/pilosa.go index e201de283..f68082985 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -32,8 +32,8 @@ import ( ) //////////////////////////////////////////////////////////////////////////////////// -// Main represents a test wrapper for server.Command. -type Main struct { +// Command represents a test wrapper for server.Command. +type Command struct { *server.Command commandOptions []server.CommandOption @@ -57,21 +57,14 @@ func OptAllowedOrigins(origins []string) server.CommandOption { } } -// GossipAddress returns the address on which gossip is listening after a Main -// has been setup. Useful to pass as a seed to other nodes when creating and -// testing clusters. -func (m *Main) GossipAddress() string { - return m.GossipTransport().URI.String() -} - -// NewMain returns a new instance of Main with a temporary data directory and random port. -func NewMain(opts ...server.CommandOption) *Main { +// NewCommand returns a new instance of Main with a temporary data directory and random port. +func NewCommand(opts ...server.CommandOption) *Command { path, err := ioutil.TempDir("", "pilosa-") if err != nil { panic(err) } - m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts} + m := &Command{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts} m.Config.DataDir = path m.Config.Bind = "http://localhost:0" m.Config.Cluster.Disabled = true @@ -92,58 +85,17 @@ func NewMain(opts ...server.CommandOption) *Main { return m } -// NewMainWithCluster returns a new instance of Main with clustering enabled. -func NewMainWithCluster(isCoordinator bool, opts ...server.CommandOption) *Main { - m := NewMain(opts...) +// NewCommandNode returns a new instance of Command with clustering enabled. +func NewCommandNode(isCoordinator bool, opts ...server.CommandOption) *Command { + m := NewCommand(opts...) m.Config.Cluster.Disabled = false m.Config.Cluster.Coordinator = isCoordinator return m } -// MustRunMainWithCluster ruturns a running array of *Main where -// all nodes are joined via memberlist (i.e. clustering enabled). -func MustRunMainWithCluster(t *testing.T, size int, opts ...[]server.CommandOption) []*Main { - ma, err := runMainWithCluster(size, opts...) - if err != nil { - t.Fatalf("new main array with cluster: %v", err) - } - return ma -} - -// runMainWithCluster runs an array of *Main where all nodes are -// joined via memberlist (i.e. clustering enabled). -func runMainWithCluster(size int, opts ...[]server.CommandOption) ([]*Main, error) { - if size == 0 { - return nil, errors.New("cluster must contain at least one node") - } - if len(opts) != size && len(opts) != 0 && len(opts) != 1 { - return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes") - } - - mains := make([]*Main, size) - var gossipSeeds = make([]string, size) - for i := 0; i < size; i++ { - var commandOpts []server.CommandOption - if len(opts) > 0 { - commandOpts = opts[i%len(opts)] - } - m := NewMainWithCluster(i == 0, commandOpts...) - m.Config.Gossip.Port = "0" - m.Config.Gossip.Seeds = gossipSeeds[:i] - - if err := m.Start(); err != nil { - return nil, errors.Wrapf(err, "Starting server %d", i) - } - gossipSeeds[i] = m.GossipTransport().URI.String() - mains[i] = m - } - - return mains, nil -} - -// MustRunMain returns a new, running Main. Panic on error. -func MustRunMain() *Main { - m := NewMain() +// MustRunCommand returns a new, running Main. Panic on error. +func MustRunCommand() *Command { + m := NewCommand() m.Config.Metric.Diagnostics = false // Disable diagnostics. if err := m.Start(); err != nil { panic(err) @@ -151,14 +103,21 @@ func MustRunMain() *Main { return m } +// GossipAddress returns the address on which gossip is listening after a Main +// has been setup. Useful to pass as a seed to other nodes when creating and +// testing clusters. +func (m *Command) GossipAddress() string { + return m.GossipTransport().URI.String() +} + // Close closes the program and removes the underlying data directory. -func (m *Main) Close() error { +func (m *Command) Close() error { defer os.RemoveAll(m.Config.DataDir) return m.Command.Close() } // Reopen closes the program and reopens it. -func (m *Main) Reopen() error { +func (m *Command) Reopen() error { if err := m.Command.Close(); err != nil { return err } @@ -180,10 +139,10 @@ func (m *Main) Reopen() error { } // URL returns the base URL string for accessing the running program. -func (m *Main) URL() string { return m.Server.URI.String() } +func (m *Command) URL() string { return m.Server.URI.String() } // Client returns a client to connect to the program. -func (m *Main) Client() *http.InternalClient { +func (m *Command) Client() *http.InternalClient { client, err := http.NewInternalClient(m.Server.URI.HostPort(), http.GetHTTPClient(nil)) if err != nil { panic(err) @@ -192,7 +151,7 @@ func (m *Main) Client() *http.InternalClient { } // Query executes a query against the program through the HTTP API. -func (m *Main) Query(index, rawQuery, query string) (string, error) { +func (m *Command) Query(index, rawQuery, query string) (string, error) { resp := MustDo("POST", m.URL()+fmt.Sprintf("/index/%s/query?", index)+rawQuery, query) if resp.StatusCode != gohttp.StatusOK { return "", fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) @@ -200,7 +159,7 @@ func (m *Main) Query(index, rawQuery, query string) (string, error) { return resp.Body, nil } -func (m *Main) RecalculateCaches() error { +func (m *Command) RecalculateCaches() error { resp := MustDo("POST", fmt.Sprintf("%s/recalculate-caches", m.URL()), "") if resp.StatusCode != 204 { return fmt.Errorf("invalid status: %d, body=%s", resp.StatusCode, resp.Body) @@ -208,6 +167,85 @@ func (m *Main) RecalculateCaches() error { return nil } +// Cluster represents a Pilosa cluster (multiple Command instances) +type Cluster []*Command + +// Start runs a Cluster +func (c Cluster) Start() error { + var gossipSeeds = make([]string, len(c)) + for i, cc := range c { + cc.Config.Gossip.Port = "0" + cc.Config.Gossip.Seeds = gossipSeeds[:i] + if err := cc.Start(); err != nil { + return errors.Wrapf(err, "starting server %d", i) + } + gossipSeeds[i] = cc.GossipTransport().URI.String() + } + return nil +} + +// Stop stops a Cluster +func (c Cluster) Close() error { + for i, cc := range c { + if err := cc.Close(); err != nil { + return errors.Wrapf(err, "stopping server %d", i) + } + } + return nil +} + +// MustNewCluster creates a new cluster +func MustNewCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster { + c, err := newCluster(size, opts...) + if err != nil { + t.Fatalf("new cluster: %v", err) + } + return c +} + +// newCluster creates a new cluster +func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) { + if size == 0 { + return nil, errors.New("cluster must contain at least one node") + } + if len(opts) != size && len(opts) != 0 && len(opts) != 1 { + return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes") + } + + cluster := make(Cluster, size) + for i := 0; i < size; i++ { + var commandOpts []server.CommandOption + if len(opts) > 0 { + commandOpts = opts[i%len(opts)] + } + m := NewCommandNode(i == 0, commandOpts...) + cluster[i] = m + } + + return cluster, nil +} + +// runCluster creates and starts a new cluster +func runCluster(size int, opts ...[]server.CommandOption) (Cluster, error) { + cluster, err := newCluster(size, opts...) + if err != nil { + return nil, errors.Wrap(err, "new cluster") + } + if err = cluster.Start(); err != nil { + return nil, errors.Wrap(err, "starting cluster") + } + return cluster, nil +} + +// MustRunCluster creates and starts a new cluster +func MustRunCluster(t *testing.T, size int, opts ...[]server.CommandOption) Cluster { + c, err := runCluster(size, opts...) + if err != nil { + t.Fatalf("run cluster: %v", err) + } + return c +} + //////////////////////////////////////////////////////////////////////////////////// // MustDo executes http.Do() with an http.NewRequest(). Panic on error. diff --git a/test/pilosa_test.go b/test/pilosa_test.go index 2ba7504c8..25bb808df 100644 --- a/test/pilosa_test.go +++ b/test/pilosa_test.go @@ -27,7 +27,7 @@ import ( func TestNewCluster(t *testing.T) { numNodes := 3 - cluster := test.MustRunMainWithCluster(t, numNodes) + cluster := test.MustRunCluster(t, numNodes) coordinator := getCoordinator(cluster[0]) for i := 1; i < numNodes; i++ { @@ -78,7 +78,7 @@ func TestNewCluster(t *testing.T) { } } -func getCoordinator(m *test.Main) string { +func getCoordinator(m *test.Command) string { hosts := m.API.Hosts(context.Background()) for _, host := range hosts { if host.IsCoordinator { From 4f3cf9af30a89e7b41a73886bb1c2d5911af61a0 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Thu, 28 Jun 2018 13:20:57 -0500 Subject: [PATCH 6/6] Use GossipAddress() helper --- test/pilosa.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/pilosa.go b/test/pilosa.go index f68082985..d2f2f6106 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -179,7 +179,7 @@ func (c Cluster) Start() error { if err := cc.Start(); err != nil { return errors.Wrapf(err, "starting server %d", i) } - gossipSeeds[i] = cc.GossipTransport().URI.String() + gossipSeeds[i] = cc.GossipAddress() } return nil }