From 2f1e3b60783cffb2ac339039979b9ae284994750 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Wed, 12 Aug 2015 14:35:24 -0600 Subject: [PATCH] Refactor config & service This commit refactors the config into a `main.Config` object instead of a global singleton. The `core.Service` is also refactored into the `main` package and individual pieces of the service are wired together by the `pilosa` binary. These two changes are required to begin to decouple packages from one another and allow them to be individually unit tested. Previously most top-level objects in the system could access any other top-level object through the `core.Service` which effectively made `Service` a singleton in the system. Each top-level object now has inline interfaces for their dependencies so that can be set at runtime by the `main` package or can be mocked by a test package. --- cmd/pilosa/config.go | 120 ++++++ cmd/pilosa/config_test.go | 190 ++++++++++ cmd/pilosa/main.go | 222 +++++++++++ cmd/pilosa/main_test.go | 1 + cmd/pilosa/seelog.go | 28 ++ commands/pilosa/main.go | 60 --- config/config.go | 165 -------- config/config_test.go | 67 ---- config/test.yaml | 3 - core/batch.go | 28 +- core/core_test.go | 14 - core/etcd.go | 102 +++-- core/http.go | 93 +++-- core/load.go | 10 +- core/ping.go | 22 +- core/query.go | 353 ----------------- core/remotebits.go | 21 +- core/service.go | 85 ----- core/topn.go | 199 ---------- db/topology.go | 8 +- default.yaml | 2 +- deps.json | 5 + dispatch/dispatch.go | 78 +++- executor/executor.go | 625 +++++++++++++++++++++++++++++-- index/bitmap_test.go | 134 +------ index/brand.go | 5 +- index/brand_test.go | 129 +------ index/fragment_container.go | 15 +- index/fragment_container_test.go | 67 ++-- index/general.go | 3 +- index/storage_cass.go | 23 +- index/storage_test.go | 67 ++-- index/timeframe_test.go | 2 + transport/http.go | 2 + transport/tcp.go | 47 ++- util/id.go | 26 +- util/statd.go | 18 +- util/storage.go | 9 - 38 files changed, 1584 insertions(+), 1464 deletions(-) create mode 100644 cmd/pilosa/config.go create mode 100644 cmd/pilosa/config_test.go create mode 100644 cmd/pilosa/main.go create mode 100644 cmd/pilosa/main_test.go create mode 100644 cmd/pilosa/seelog.go delete mode 100644 commands/pilosa/main.go delete mode 100644 config/config.go delete mode 100644 config/config_test.go delete mode 100644 config/test.yaml delete mode 100644 core/core_test.go delete mode 100644 core/query.go delete mode 100644 core/topn.go diff --git a/cmd/pilosa/config.go b/cmd/pilosa/config.go new file mode 100644 index 000000000..4ddc7484e --- /dev/null +++ b/cmd/pilosa/config.go @@ -0,0 +1,120 @@ +package main + +import ( + "time" + + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/transport" + "github.com/umbel/pilosa/util" +) + +const ( + // DefaultLogPath is the default file path where the log will be written. + DefaultLogPath = "/tmps" + + // DefaultLogLevel is the default logging level used by seelog. + DefaultLogLevel = "info" + + // DefaultFragmentBase is the default path where fragments are stored. + DefaultFragmentBase = "/tmp/single" +) + +var ( + // DefaultSupportedFrames are the frames supported by default. + DefaultSupportedFrames = [...]string{"b.n", "t.t", "l.n", "d", "p.n"} + + // DefaultETCDHosts are the default hosts to connect to. + DefaultETCDHosts = [...]string{"http://127.0.0.1:4001"} +) + +// Config represents the configuration for the command. +type Config struct { + ID *util.GUID `toml:"id"` + Host string `toml:"host"` + + TCP struct { + Port int `toml:"port"` + } `toml:"tcp"` + + HTTP struct { + Port int `toml:"port"` + DefaultDB string `toml:"default-db"` + RequestLogPath string `toml:"request-log-path"` + SetBitLogEnabled bool `toml:"set-bit-log-enabled"` + } `toml:"http"` + + Log struct { + Path string `toml:"path"` + Level string `toml:"level"` + } + + Storage struct { + Backend string `toml:"backend"` + Hosts []string `toml:"hosts"` + Keyspace string `toml:"keyspace"` + + FragmentBase string `toml:"fragment-base"` + SupportedFrames []string `toml:"supported-frames"` + + CassandraTimeWindow Duration `toml:"cassandra-time-window"` + CassandraMaxSizeBatch int `toml:"cassandra-max-size-batch"` + } `toml:"storage"` + + AWS struct { + AccessKeyID string `toml:"access-key-id"` + SecretAccessKey string `toml:"secret-access-key"` + } `toml:"aws"` + + LevelDB struct { + Path string `toml:"path"` + } `toml:"leveldb"` + + Statsd struct { + Host string `toml:"host"` + } `toml:"statsd"` + + Plugins struct { + Path string `toml:"path"` + } `toml:"plugins"` + + ETCD struct { + Hosts []string `toml:"hosts"` + FragmentAllocLockTTL Duration `toml:"fragment-alloc-lock-ttl"` + } `toml:"etcd"` +} + +// NewConfig returns an instance of Config with default options. +func NewConfig() Config { + var c Config + c.Host = "localhost" + c.TCP.Port = transport.DefaultTCPPort + c.HTTP.Port = transport.DefaultHTTPPort + c.Log.Path = DefaultLogPath + c.Storage.Backend = index.DefaultBackend + c.Storage.Hosts = index.DefaultStorageHosts[:] + c.Storage.Keyspace = index.DefaultStorageKeyspace + c.Storage.FragmentBase = DefaultFragmentBase + c.Storage.SupportedFrames = DefaultSupportedFrames[:] + c.Storage.CassandraTimeWindow = Duration(index.DefaultCassandraTimeWindow) + c.Storage.CassandraMaxSizeBatch = index.DefaultCassandraMaxSizeBatch + c.Statsd.Host = util.DefaultStatsdHost + c.ETCD.Hosts = DefaultETCDHosts[:] + return c +} + +// Duration is a TOML wrapper type for time.Duration. +type Duration time.Duration + +// String returns the string representation of the duration. +func (d Duration) String() string { return time.Duration(d).String() } + +// UnmarshalText parses a TOML value into a duration value. +func (d *Duration) UnmarshalText(text []byte) error { + v, err := time.ParseDuration(string(text)) + if err != nil { + return err + } + + *d = Duration(v) + return nil +} diff --git a/cmd/pilosa/config_test.go b/cmd/pilosa/config_test.go new file mode 100644 index 000000000..3ea0fda09 --- /dev/null +++ b/cmd/pilosa/config_test.go @@ -0,0 +1,190 @@ +package main_test + +import ( + "reflect" + "testing" + "time" + + "github.com/BurntSushi/toml" + "github.com/umbel/pilosa/cmd/pilosa" +) + +// Ensure the ID can be parsed as a GUID. +func TestConfig_Parse_ID(t *testing.T) { + if c, err := ParseConfig(` +id = "00000000-0000-0000-0000-000000000001" +`); err != nil { + t.Fatal(err) + } else if c.ID == nil || c.ID.String() != `00000000-0000-0000-0000-000000000001` { + t.Fatalf("unexpected id: %s", c.ID) + } +} + +// Ensure that parsing an invalid GUID returns an error. +func TestConfig_Parse_ID_ErrInvalid(t *testing.T) { + if _, err := ParseConfig(` +id = "x" +`); err == nil || err.Error() != `Type mismatch for 'main.Config.id': invalid GUID "x"` { + t.Fatal(err) + } +} + +// Ensure the host can be parsed. +func TestConfig_Parse_Host(t *testing.T) { + if c, err := ParseConfig(`host = "localhost"`); err != nil { + t.Fatal(err) + } else if c.Host != "localhost" { + t.Fatalf("unexpected host: %s", c.Host) + } +} + +// Ensure the "tcp" config can be parsed. +func TestConfig_Parse_TCP(t *testing.T) { + if c, err := ParseConfig(` +[tcp] +port = 123 +`); err != nil { + t.Fatal(err) + } else if c.TCP.Port != 123 { + t.Fatalf("unexpected port: %s", c.TCP.Port) + } +} + +// Ensure the "http" config can be parsed. +func TestConfig_Parse_HTTP(t *testing.T) { + if c, err := ParseConfig(` +[http] +port = 123 +default-db = "xyz" +request-log-path = "/path/to/log" +set-bit-log-enabled = true +`); err != nil { + t.Fatal(err) + } else if c.HTTP.Port != 123 { + t.Fatalf("unexpected port: %s", c.HTTP.Port) + } else if c.HTTP.DefaultDB != "xyz" { + t.Fatalf("unexpected default db: %s", c.HTTP.DefaultDB) + } else if c.HTTP.RequestLogPath != "/path/to/log" { + t.Fatalf("unexpected request log path: %s", c.HTTP.RequestLogPath) + } else if c.HTTP.SetBitLogEnabled != true { + t.Fatalf("unexpected set bit log enabled: %v", c.HTTP.SetBitLogEnabled) + } +} + +// Ensure the "log" config can be parsed. +func TestConfig_Parse_Log(t *testing.T) { + if c, err := ParseConfig(` +[log] +path = "/path/to/log" +level = "debug" +`); err != nil { + t.Fatal(err) + } else if c.Log.Path != "/path/to/log" { + t.Fatalf("unexpected path: %s", c.Log.Path) + } else if c.Log.Level != "debug" { + t.Fatalf("unexpected level: %s", c.Log.Level) + } +} + +// Ensure the "storage" config can be parsed. +func TestConfig_Parse_Storage(t *testing.T) { + if c, err := ParseConfig(` +[storage] +backend = "cassandra" +hosts = ["server0", "server1"] +keyspace = "pilosa" +fragment-base = "/path/to/base" +supported-frames = ["a", "b", "c"] +cassandra-time-window = "5s" +cassandra-max-size-batch = 50 +`); err != nil { + t.Fatal(err) + } else if c.Storage.Backend != "cassandra" { + t.Fatalf("unexpected backend: %s", c.Storage.Backend) + } else if !reflect.DeepEqual(c.Storage.Hosts, []string{"server0", "server1"}) { + t.Fatalf("unexpected hosts: %+v", c.Storage.Hosts) + } else if c.Storage.Keyspace != "pilosa" { + t.Fatalf("unexpected keyspace: %s", c.Storage.Keyspace) + } else if c.Storage.FragmentBase != "/path/to/base" { + t.Fatalf("unexpected fragment base: %s", c.Storage.FragmentBase) + } else if !reflect.DeepEqual(c.Storage.SupportedFrames, []string{"a", "b", "c"}) { + t.Fatalf("unexpected supported frames: %s", c.Storage.SupportedFrames) + } else if time.Duration(c.Storage.CassandraTimeWindow) != 5*time.Second { + t.Fatalf("unexpected cassandra time window: %s", time.Duration(c.Storage.CassandraTimeWindow)) + } else if c.Storage.CassandraMaxSizeBatch != 50 { + t.Fatalf("unexpected cassandra max size batch: %s", c.Storage.CassandraMaxSizeBatch) + } +} + +// Ensure the "aws" config can be parsed. +func TestConfig_Parse_AWS(t *testing.T) { + if c, err := ParseConfig(` +[aws] +access-key-id = "abc" +secret-access-key = "def" +`); err != nil { + t.Fatal(err) + } else if c.AWS.AccessKeyID != "abc" { + t.Fatalf("unexpected access key id: %s", c.AWS.AccessKeyID) + } else if c.AWS.SecretAccessKey != "def" { + t.Fatalf("unexpected secret access key: %s", c.AWS.SecretAccessKey) + } +} + +// Ensure the "leveldb" config can be parsed. +func TestConfig_Parse_LevelDB(t *testing.T) { + if c, err := ParseConfig(` +[leveldb] +path = "/path/to/db" +`); err != nil { + t.Fatal(err) + } else if c.LevelDB.Path != "/path/to/db" { + t.Fatalf("unexpected path: %s", c.LevelDB.Path) + } +} + +// Ensure the "statsd" config can be parsed. +func TestConfig_Parse_Statsd(t *testing.T) { + if c, err := ParseConfig(` +[statsd] +host = "localhost" +`); err != nil { + t.Fatal(err) + } else if c.Statsd.Host != "localhost" { + t.Fatalf("unexpected host: %s", c.Statsd.Host) + } +} + +// Ensure the "plugins" config can be parsed. +func TestConfig_Parse_Plugins(t *testing.T) { + if c, err := ParseConfig(` +[plugins] +path = "/path/to/plugins" +`); err != nil { + t.Fatal(err) + } else if c.Plugins.Path != "/path/to/plugins" { + t.Fatalf("unexpected path: %s", c.Plugins.Path) + } +} + +// Ensure the "etcd" config can be parsed. +func TestConfig_Parse_ETCD(t *testing.T) { + if c, err := ParseConfig(` +[etcd] +hosts = ["127.0.0.1"] +fragment-alloc-lock-ttl = "5m" +`); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(c.ETCD.Hosts, []string{"127.0.0.1"}) { + t.Fatalf("unexpected hosts: %+v", c.ETCD.Hosts) + } else if time.Duration(c.ETCD.FragmentAllocLockTTL) != 5*time.Minute { + t.Fatalf("unexpected fragment alloc lock ttl: %v", c.ETCD.FragmentAllocLockTTL) + } +} + +// ParseConfig parses s into a config. +func ParseConfig(s string) (main.Config, error) { + var c main.Config + _, err := toml.Decode(s, &c) + return c, err +} diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go new file mode 100644 index 000000000..79a452381 --- /dev/null +++ b/cmd/pilosa/main.go @@ -0,0 +1,222 @@ +package main + +import ( + "flag" + "fmt" + "io" + "os" + "runtime/pprof" + "time" + + "github.com/BurntSushi/toml" + log "github.com/cihub/seelog" + "github.com/coreos/go-etcd/etcd" + "github.com/kr/s3/s3util" + "github.com/mitchellh/panicwrap" + "github.com/umbel/pilosa/core" + "github.com/umbel/pilosa/db" + "github.com/umbel/pilosa/dispatch" + "github.com/umbel/pilosa/executor" + "github.com/umbel/pilosa/hold" + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/transport" + "github.com/umbel/pilosa/util" +) + +// Build holds the build information passed in at compile time. +var Build string + +func main() { + m := NewMain() + if err := m.Run(os.Args[1:]...); err != nil { + fmt.Fprintln(m.Stderr, err.Error()) + os.Exit(-1) + } +} + +// Main represents the main program execution. +type Main struct { + Stdout io.Writer + Stderr io.Writer +} + +// NewMain returns a new instance of Main. +func NewMain() *Main { + return &Main{ + Stdout: os.Stdout, + Stderr: os.Stderr, + } +} + +// Run executes the main program execution. +func (m *Main) Run(args ...string) error { + defer log.Flush() + + // Handle panics with a log entry and process exit. + if code, err := panicwrap.BasicWrap(func(output string) { + log.Critical("The child panicked:\n\n", output) + os.Exit(1) + }); err != nil { + panic(err) + } else if code >= 0 { + os.Exit(code) + } + + // Parse command line arguments. + opt, err := m.ParseFlags(args) + if err != nil { + return err + } + + // Parse configuration. + config := NewConfig() + if opt.ConfigPath != "" { + if _, err := toml.DecodeFile(opt.ConfigPath, &config); err != nil { + return err + } + } + + // Generate an ID if one is not specified in the config. + id := config.ID + if id == nil { + *id = util.RandomUUID() + } + + // Set up profiling. + if opt.CPUProfile != "" { + f, err := os.Create(opt.CPUProfile) + if err != nil { + return err + } + + pprof.StartCPUProfile(f) + defer pprof.StopCPUProfile() + } + + // Pass configuration to packages. + // NOTE: This is temporary. These config options should be encapsulated in the types. + db.SupportedFrames = config.Storage.SupportedFrames + index.FragmentBase = config.Storage.FragmentBase + index.Backend = config.Storage.Backend + index.LevelDBPath = config.LevelDB.Path + + // Initialize AWS storage. + s3util.DefaultConfig.AccessKey = config.AWS.AccessKeyID + s3util.DefaultConfig.SecretKey = config.AWS.SecretAccessKey + + // Initialize Statsd. + util.StatsdHost = config.Statsd.Host + util.SetupStatsd() + + // REMOVED(benbjohnson): config.SetupConfig() + index.SetupCassandra() + + // Initialize logging. + logger, _ := log.LoggerFromConfigAsBytes([]byte(SeelogProductionConfig(config.Log.Path, *id, config.Log.Level))) + log.ReplaceLogger(logger) + + // Initialize etcd client. + etcdClient := etcd.NewClient(config.ETCD.Hosts) + + // Initialize the cluster. + cluster := db.NewCluster() + + // Create index. + idx := index.NewFragmentContainer() + + // Initialize the holder. + hold := hold.NewHolder() + + // Initialize process map. + processMap := core.NewProcessMap() + + // Start process mapper. + processMapper := core.NewProcessMapper("/pilosa/0") + processMapper.ID = *id + processMapper.TCPPort = config.TCP.Port + processMapper.HTTPPort = config.HTTP.Port + processMapper.Host = config.Host + processMapper.ProcessMap = processMap + processMapper.EtcdClient = etcdClient + + // Start topology mapper. + topologyMapper := core.NewTopologyMapper("/pilosa/0") + topologyMapper.Cluster = cluster + topologyMapper.ProcessMap = processMap + topologyMapper.EtcdClient = etcdClient + topologyMapper.Index = idx + topologyMapper.SupportedFrames = config.Storage.SupportedFrames + topologyMapper.FragmentAllocLockTTL = time.Duration(config.ETCD.FragmentAllocLockTTL) + + // Start the transport. + transport := transport.NewTcpTransport(*id) + transport.Port = config.TCP.Port + transport.ProcessMap = processMap + go transport.Run() + + // Create the pinger. + pinger := core.NewPinger(*id) + pinger.Hold = hold + pinger.Transport = transport + + // Create the batcher. + batcher := core.NewBatcher(*id) + batcher.Cluster = cluster + batcher.Hold = hold + batcher.Transport = transport + + // Start the web service. + core.RequestLogPath = config.HTTP.RequestLogPath + ws := core.NewWebService() + ws.ID = *id + ws.Port = config.HTTP.Port + ws.Version = Build + ws.DefaultDB = config.HTTP.DefaultDB + ws.SetBitLogEnabled = config.HTTP.SetBitLogEnabled + ws.Cluster = cluster + ws.TopologyMapper = topologyMapper + ws.Pinger = pinger + ws.Batcher = batcher + + // Start the executor. + ex := executor.NewExecutor(*id) + ex.ProcessMap = processMap + ex.PluginsPath = config.Plugins.Path + ex.Hold = hold + ex.Index = idx + + // Start the dispatcher. + dispatch := dispatch.NewDispatch() + dispatch.Executor = ex + dispatch.Hold = hold + dispatch.Index = idx + dispatch.Transport = transport + go dispatch.Run() + + fmt.Printf("Pilosa %s\n", Build) + + log.Warn("STOP") + + return nil +} + +// ParseFlags parses command line flags from args. +func (m *Main) ParseFlags(args []string) (Options, error) { + var opt Options + fs := flag.NewFlagSet("pilosa", flag.ContinueOnError) + fs.SetOutput(m.Stderr) + fs.StringVar(&opt.ConfigPath, "config", "", "config path") + fs.StringVar(&opt.CPUProfile, "cpuprofile", "", "write cpu profile to file") + + if err := fs.Parse(args); err != nil { + return opt, err + } + + return opt, nil +} + +// Options represents the command line options. +type Options struct { + CPUProfile string + ConfigPath string +} diff --git a/cmd/pilosa/main_test.go b/cmd/pilosa/main_test.go new file mode 100644 index 000000000..0fee6f5dc --- /dev/null +++ b/cmd/pilosa/main_test.go @@ -0,0 +1 @@ +package main_test diff --git a/cmd/pilosa/seelog.go b/cmd/pilosa/seelog.go new file mode 100644 index 000000000..6b0112390 --- /dev/null +++ b/cmd/pilosa/seelog.go @@ -0,0 +1,28 @@ +package main + +import ( + "fmt" + + "github.com/umbel/pilosa/util" +) + +func SeelogProductionConfig(path string, id util.GUID, level string) string { + if path == "" { + path = "/tmp" + } + + fname := fmt.Sprintf("%s/pilosa.%s", path, id.String()) + // + + s := fmt.Sprintf(` + + + `, level, fname) + + s += ` + + + ` + + return s +} diff --git a/commands/pilosa/main.go b/commands/pilosa/main.go deleted file mode 100644 index 510cfde05..000000000 --- a/commands/pilosa/main.go +++ /dev/null @@ -1,60 +0,0 @@ -package main - -import ( - "flag" - "os" - "runtime/pprof" - - log "github.com/cihub/seelog" - "github.com/mitchellh/panicwrap" - "github.com/umbel/pilosa/config" - "github.com/umbel/pilosa/core" - "github.com/umbel/pilosa/cruncher" - "github.com/umbel/pilosa/index" - "github.com/umbel/pilosa/util" -) - -var ( - cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file") - Build string -) - -func main() { - defer log.Flush() - exitStatus, err := panicwrap.BasicWrap(panicHandler) - if err != nil { - // Something went wrong setting up the panic wrapper. Unlikely, - // but possible. - panic(err) - } - - if exitStatus >= 0 { - os.Exit(exitStatus) - } - core.Build = Build - - flag.Parse() - if *cpuprofile != "" { - f, err := os.Create(*cpuprofile) - if err != nil { - log.Warn(err) - os.Exit(1) - } - pprof.StartCPUProfile(f) - defer pprof.StopCPUProfile() - } - config.SetupConfig() - util.SetupUtil() - index.SetupCassandra() - - cruncher := cruncher.NewCruncher() - cruncher.Run() - log.Warn("STOP") -} - -func panicHandler(output string) { - // output contains the full output (including stack traces) of the - // panic. Put it in a file or something. - log.Critical("The child panicked:\n\n", output) - os.Exit(1) -} diff --git a/config/config.go b/config/config.go deleted file mode 100644 index 0c117428e..000000000 --- a/config/config.go +++ /dev/null @@ -1,165 +0,0 @@ -package config - -import ( - "errors" - "io/ioutil" - "os" - "sync" - - log "github.com/cihub/seelog" - "launchpad.net/goyaml" -) - -type Config struct { - config map[string]interface{} - lock sync.RWMutex - filename string - loaded bool -} - -var config *Config - -func SetupConfig() { - config = NewConfig("") -} - -func GetSafe(key string) (interface{}, bool) { - return config.GetSafe(key) -} - -func Get(key string) interface{} { - return config.Get(key) -} - -func GetInt(key string) int { - return config.GetInt(key) -} - -func GetString(key string) string { - return config.GetString(key) -} - -func GetStringArray(key string) []string { - res, _ := config.GetStringArray(key) - return res -} -func GetStringArrayDefault(key string, def []string) []string { - res, ok := config.GetStringArray(key) - - if !ok { - return def - } - return res -} - -func GetStringDefault(key string, default_value string) string { - return config.GetStringDefault(key, default_value) -} - -func GetIntDefault(key string, default_value int) int { - return config.GetIntDefault(key, default_value) -} - -func NewConfig(filename string) *Config { - self := Config{} - self.config = make(map[string]interface{}) - self.filename = filename - return &self -} - -func (self *Config) load() error { - self.lock.Lock() - defer self.lock.Unlock() - if self.loaded { - return nil - } - config_file := self.filename - if config_file == "" { - config_file = os.Getenv("PILOSA_CONFIG") - if config_file == "" { - log.Warn("PILOSA_CONFIG not set, defaulting to pilosa.yaml") - config_file = "pilosa.yaml" - } - } - data, err := ioutil.ReadFile(config_file) - if err != nil { - return errors.New("Problem with config file: " + err.Error()) - } - err = goyaml.Unmarshal(data, self.config) - if err != nil { - println(err.Error()) - - } - self.loaded = true - return nil -} - -func (self *Config) GetSafe(key string) (interface{}, bool) { - self.load() - self.lock.RLock() - defer self.lock.RUnlock() - value, ok := self.config[key] - return value, ok -} - -func (self *Config) Get(key string) interface{} { - self.load() - self.lock.RLock() - defer self.lock.RUnlock() - return self.config[key] -} - -func (self *Config) GetInt(key string) int { - value, ok := self.GetSafe(key) - if ok { - value_int, ok := value.(int) - if ok { - return value_int - } - } - return 0 -} -func (self *Config) GetIntDefault(key string, default_value int) int { - value, ok := self.GetSafe(key) - if ok { - value_int, ok := value.(int) - if ok { - return value_int - } - } - return default_value -} - -func (self *Config) GetStringDefault(key string, default_value string) string { - value, ok := self.GetSafe(key) - if ok { - value_string, ok := value.(string) - if ok { - return value_string - } - } - return default_value -} - -func (self *Config) GetString(key string) string { - value, ok := self.GetSafe(key) - if ok { - value_string, ok := value.(string) - if ok { - return value_string - } - } - return "" -} - -func (self *Config) GetStringArray(key string) ([]string, bool) { - value, ok := self.GetSafe(key) - if ok { - var results []string - for _, v := range value.([]interface{}) { - results = append(results, v.(string)) - } - return results, ok - } - return []string{}, ok -} diff --git a/config/config_test.go b/config/config_test.go deleted file mode 100644 index 695f06399..000000000 --- a/config/config_test.go +++ /dev/null @@ -1,67 +0,0 @@ -package config - -import ( - "os" - "testing" - - . "github.com/smartystreets/goconvey/convey" -) - -func TestConfig(t *testing.T) { - err := os.Setenv("PILOSA_CONFIG", "test.yaml") - if err != nil { - t.Fatal("Error setting PILOSA_CONFIG") - } - Convey("config.Get()", t, func() { - So(Get("port_tcp"), ShouldEqual, 12000) - So(Get("port_http"), ShouldEqual, 15000) - So(Get("temp"), ShouldEqual, "/tmp") - So(Get("notfound"), ShouldBeNil) - }) - Convey("config.GetSafe()", t, func() { - val, ok := GetSafe("port_tcp") - So(ok, ShouldBeTrue) - So(val, ShouldEqual, 12000) - val, ok = GetSafe("derp") - So(ok, ShouldBeFalse) - So(val, ShouldBeNil) - }) - Convey("config.GetInt()", t, func() { - So(GetInt("port_tcp"), ShouldEqual, 12000) - So(GetInt("port_http"), ShouldEqual, 15000) - So(GetInt("notfound"), ShouldEqual, 0) - }) - Convey("config.GetString()", t, func() { - So(GetString("temp"), ShouldEqual, "/tmp") - }) -} - -func TestConfigObject(t *testing.T) { - err := os.Setenv("PILOSA_CONFIG", "") - if err != nil { - t.Fatal("Error setting PILOSA_CONFIG") - } - conf := NewConfig("test.yaml") - Convey("config.Get()", t, func() { - So(conf.Get("port_tcp"), ShouldEqual, 12000) - So(conf.Get("port_http"), ShouldEqual, 15000) - So(conf.Get("temp"), ShouldEqual, "/tmp") - So(conf.Get("notfound"), ShouldBeNil) - }) - Convey("config.GetSafe()", t, func() { - val, ok := conf.GetSafe("port_tcp") - So(ok, ShouldBeTrue) - So(val, ShouldEqual, 12000) - val, ok = conf.GetSafe("derp") - So(ok, ShouldBeFalse) - So(val, ShouldBeNil) - }) - Convey("config.GetInt()", t, func() { - So(conf.GetInt("port_tcp"), ShouldEqual, 12000) - So(conf.GetInt("port_http"), ShouldEqual, 15000) - So(conf.GetInt("notfound"), ShouldEqual, 0) - }) - Convey("config.GetString()", t, func() { - So(conf.GetString("temp"), ShouldEqual, "/tmp") - }) -} diff --git a/config/test.yaml b/config/test.yaml deleted file mode 100644 index 32bc32da8..000000000 --- a/config/test.yaml +++ /dev/null @@ -1,3 +0,0 @@ -port_tcp: 12000 -port_http: 15000 -temp: /tmp diff --git a/core/batch.go b/core/batch.go index c577c407c..618895150 100644 --- a/core/batch.go +++ b/core/batch.go @@ -5,6 +5,7 @@ import ( log "github.com/cihub/seelog" "github.com/umbel/pilosa/db" + "github.com/umbel/pilosa/hold" "github.com/umbel/pilosa/util" ) @@ -33,22 +34,37 @@ func init() { gob.Register(BatchResponse{}) } -func (self *Service) Batch(database_name, frame, compressed_bitmap string, bitmap_id uint64, slice int, filter uint64) error { +type Batcher struct { + ID util.GUID + Cluster *db.Cluster + Hold *hold.Holder + + Transport interface { + Send(message *db.Message, host *util.GUID) + } +} + +func NewBatcher(id util.GUID) *Batcher { + return &Batcher{ID: id} +} + +func (b *Batcher) Batch(database_name, frame, compressed_bitmap string, bitmap_id uint64, slice int, filter uint64) error { log.Trace("Batch:", "db:", database_name, " frame:", frame, " slice:", slice, " cb:", compressed_bitmap, " bid:", bitmap_id, "f:", filter) + //determine the fragment_id from database/frame/slice - database := self.Cluster.GetOrCreateDatabase(database_name) + database := b.Cluster.GetOrCreateDatabase(database_name) oslice := database.GetOrCreateSlice(slice) //need to find processid and fragment id for that slice fragment, err := database.GetFragmentForBitmap(oslice, &db.Bitmap{bitmap_id, frame, filter}) if err == nil { id := util.RandomUUID() - batch := db.Message{Data: BatchRequest{Id: &id, Source: self.Id, Fragment_id: fragment.GetId(), Bitmap_id: bitmap_id, Compressed_bitmap: compressed_bitmap}} + batch := db.Message{Data: BatchRequest{Id: &id, Source: &b.ID, Fragment_id: fragment.GetId(), Bitmap_id: bitmap_id, Compressed_bitmap: compressed_bitmap}} dest_id := fragment.GetProcess().Id() - self.Transport.Send(&batch, &dest_id) + b.Transport.Send(&batch, &dest_id) - _, err = self.Hold.Get(&id, 60) + _, err = b.Hold.Get(&id, 60) } - return err + return err } diff --git a/core/core_test.go b/core/core_test.go deleted file mode 100644 index d48ca3fea..000000000 --- a/core/core_test.go +++ /dev/null @@ -1,14 +0,0 @@ -package core - -import ( - "testing" - - . "github.com/smartystreets/goconvey/convey" -) - -func TestCompile(t *testing.T) { - Convey("has asm", t, func() { - - So(canCompile(), ShouldEqual, true) - }) -} diff --git a/core/etcd.go b/core/etcd.go index c4a0a3ea7..a42cb9062 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -9,28 +9,57 @@ import ( "strconv" "strings" "sync" + "time" log "github.com/cihub/seelog" "github.com/coreos/go-etcd/etcd" "github.com/davecgh/go-spew/spew" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/db" "github.com/umbel/pilosa/util" ) +const ( + DefaultFragmentAllocLockTTL = 14400 * time.Second +) + type TopologyMapper struct { - service *Service namespace string + + ID util.GUID + Cluster *db.Cluster + ProcessMap *ProcessMap + + SupportedFrames []string + FragmentAllocLockTTL time.Duration + + EtcdClient interface { + CreateDir(key string, ttl uint64) (*etcd.Response, error) + Get(key string, sort, recursive bool) (*etcd.Response, error) + RawCreate(key string, value string, ttl uint64) (*etcd.RawResponse, error) + Set(key string, value string, ttl uint64) (*etcd.Response, error) + Watch(prefix string, waitIndex uint64, recursive bool, receiver chan *etcd.Response, stop chan bool) (*etcd.Response, error) + } + + Index interface { + AddFragment(db string, frame string, slice int, id util.SUUID) + } +} + +func NewTopologyMapper(namespace string) *TopologyMapper { + return &TopologyMapper{ + namespace: namespace, + FragmentAllocLockTTL: DefaultFragmentAllocLockTTL, + } } func (self *TopologyMapper) Setup() { log.Warn(self.namespace + "/db") db_path := self.namespace + "/db" - resp, err := self.service.Etcd.Get(db_path, false, true) + resp, err := self.EtcdClient.Get(db_path, false, true) if err != nil { ee, ok := err.(*etcd.EtcdError) if ok && ee.ErrorCode == 100 { // node does not exist - resp, err = self.service.Etcd.CreateDir(db_path, 0) + resp, err = self.EtcdClient.CreateDir(db_path, 0) if err != nil { log.Critical(err) os.Exit(-1) @@ -40,6 +69,7 @@ func (self *TopologyMapper) Setup() { os.Exit(-1) } } + //need to lock the world for _, node := range flatten(resp.Node) { err := self.handlenode(node) @@ -59,7 +89,7 @@ func (self *TopologyMapper) Run() { ns := self.namespace + "/db" log.Warn(" ETCD watcher:", ns) stop := make(chan bool) - resp, err := self.service.Etcd.Watch(ns, 0, true, receiver, stop) + resp, err := self.EtcdClient.Watch(ns, 0, true, receiver, stop) log.Warn("TopologyMapper ETCD watcher", resp, err) } }() @@ -76,10 +106,6 @@ func (self *TopologyMapper) Run() { }() } -func NewTopologyMapper(service *Service, namespace string) *TopologyMapper { - return &TopologyMapper{service, namespace} -} - type Pair struct { Key string Value int @@ -112,14 +138,13 @@ func getLightestProcess(m map[string]int) (Pair, error) { func (self *TopologyMapper) GetProcessFragmentCounts() map[string]int { m := make(map[string]int) - id_string := self.service.Id.String() - m[id_string] = 0 //at least have one process if none created - for k, _ := range self.service.ProcessMap.nodes { + m[self.ID.String()] = 0 //at least have one process if none created + for k, _ := range self.ProcessMap.nodes { p := k.String() m[p] = 0 //at least have one process if none created - } - for _, dbs := range self.service.Cluster.GetDatabases() { + + for _, dbs := range self.Cluster.GetDatabases() { for _, fsi := range dbs.GetFrameSliceIntersects() { for _, fragment := range fsi.GetFragments() { process := fragment.GetProcess().Id().String() @@ -137,10 +162,8 @@ func (self *TopologyMapper) GetProcessFragmentCounts() map[string]int { } func (self *TopologyMapper) MakeFragments(db string, slice_int int) error { - ttl := uint64(config.GetIntDefault("fragment_alloc_lock_time_secs", 14400)) - //lock_key := fmt.Sprintf("%s/lock/%s-%s-%d", self.namespace, db, frame, slice_int) lock_key := fmt.Sprintf("%s/lock/%s-%d", self.namespace, db, slice_int) - response, err := self.service.Etcd.RawCreate(lock_key, "0", ttl) + response, err := self.EtcdClient.RawCreate(lock_key, "0", uint64(self.FragmentAllocLockTTL.Seconds())) if err == nil { if response.StatusCode == 201 { //key created @@ -153,8 +176,7 @@ func (self *TopologyMapper) MakeFragments(db string, slice_int int) error { } // PUT -d "value=5cb315c3-6e1d-4218-89b7-943d1dba985b" http://etcd0:4001/v2/keys/pilosa/0/db/29/frame/d/slice/5/fragment/a2b632fc4001b817/proces - frames_to_create := config.GetStringArrayDefault("supported_frames", []string{"default"}) - for _, frame := range frames_to_create { + for _, frame := range self.SupportedFrames { err := self.AllocateFragment(p.Key, db, frame, slice_int) if err != nil { log.Warn(err) @@ -176,7 +198,7 @@ func (self *TopologyMapper) AllocateFragment(process_guid, db, frame string, sli // need to check value to see how many we have left log.Warn("ALLOC:", process_guid, len(process_guid)) if len(process_guid) > 1 { - _, err := self.service.Etcd.Set(fragment_key, process_guid, 0) + _, err := self.EtcdClient.Set(fragment_key, process_guid, 0) if err != nil { return err } @@ -218,7 +240,7 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { return nil } if len(bits) > 1 { - database = self.service.Cluster.GetOrCreateDatabase(bits[1]) + database = self.Cluster.GetOrCreateDatabase(bits[1]) } if len(bits) > 2 { if bits[2] != "frame" { @@ -262,8 +284,8 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { process = db.NewProcess(&process_uuid) fragment.SetProcess(process) - if util.Equal(self.service.Id, &process_uuid) { - self.service.Index.AddFragment(bits[1], bits[3], slice_int, fragment_id) + if util.Equal(&self.ID, &process_uuid) { + self.Index.AddFragment(bits[1], bits[3], slice_int, fragment_id) } } @@ -277,7 +299,7 @@ func (self *TopologyMapper) remove_fragment(node *etcd.Node) error { func flatten(node *etcd.Node) []*etcd.Node { nodes := []*etcd.Node{node} for i := 0; i < len(node.Nodes); i++ { - nodes = append(nodes, flatten(node.Nodes[i])...) + nodes = append(nodes, flatten(&node.Nodes[i])...) } return nodes } @@ -374,15 +396,26 @@ func (self *ProcessMap) GetMetadata() map[string]map[string]interface{} { } type ProcessMapper struct { - service *Service receiver chan etcd.Response commands chan ProcessMapperCommand namespace string + + ID util.GUID + ProcessMap *ProcessMap + + TCPPort int + HTTPPort int + Host string + + EtcdClient interface { + Get(key string, sort, recursive bool) (*etcd.Response, error) + Set(key string, value string, ttl uint64) (*etcd.Response, error) + Watch(prefix string, waitIndex uint64, recursive bool, receiver chan *etcd.Response, stop chan bool) (*etcd.Response, error) + } } -func NewProcessMapper(service *Service, namespace string) *ProcessMapper { +func NewProcessMapper(namespace string) *ProcessMapper { return &ProcessMapper{ - service: service, receiver: make(chan etcd.Response), commands: make(chan ProcessMapperCommand), namespace: namespace, @@ -423,7 +456,7 @@ func (self *ProcessMapper) handlenode(node *etcd.Node) error { if err != nil { return errors.New("Invalid GUID: " + id_string + " (" + key + ")") } - process = self.service.ProcessMap.GetOrAddProcess(&id) + process = self.ProcessMap.GetOrAddProcess(&id) } if len(bits) >= 3 { switch bits[2] { @@ -443,22 +476,21 @@ func (self *ProcessMapper) handlenode(node *etcd.Node) error { } func (self *ProcessMapper) Run() { - id_string := self.service.Id.String() path := self.namespace + "/process" - self_path := path + "/" + id_string + self_path := path + "/" + self.ID.String() log.Warn("Writing configuration to etcd...") log.Warn(self_path) var err error - _, err = self.service.Etcd.Set(self_path+"/port_tcp", strconv.Itoa(config.GetInt("port_tcp")), 0) + _, err = self.EtcdClient.Set(self_path+"/port_tcp", strconv.Itoa(self.TCPPort), 0) crash_on_error(err) - _, err = self.service.Etcd.Set(self_path+"/port_http", strconv.Itoa(config.GetInt("port_http")), 0) + _, err = self.EtcdClient.Set(self_path+"/port_http", strconv.Itoa(self.HTTPPort), 0) crash_on_error(err) - _, err = self.service.Etcd.Set(self_path+"/host", config.GetString("host"), 0) + _, err = self.EtcdClient.Set(self_path+"/host", self.Host, 0) crash_on_error(err) - response, err := self.service.Etcd.Get(path, false, true) + response, err := self.EtcdClient.Get(path, false, true) for _, node := range flatten(response.Node) { err := self.handlenode(node) if err != nil { @@ -472,7 +504,7 @@ func (self *ProcessMapper) Run() { go func() { // TODO: error check and restart watcher // TODO: use modindex to make sure watch catches everything - _, _ = self.service.Etcd.Watch(path, 0, true, receiver, stop) + _, _ = self.EtcdClient.Watch(path, 0, true, receiver, stop) }() go func() { diff --git a/core/http.go b/core/http.go index 2924cb2d7..c63058221 100644 --- a/core/http.go +++ b/core/http.go @@ -20,20 +20,50 @@ import ( log "github.com/cihub/seelog" "github.com/davecgh/go-spew/spew" "github.com/gorilla/websocket" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/db" "github.com/umbel/pilosa/index" "github.com/umbel/pilosa/util" ) +const DefaultRequestLogPath = "/tmp/set_bit_log" + +var RequestLogPath = DefaultRequestLogPath + type WebService struct { - service *Service - end chan bool + end chan bool + + ID util.GUID + Version string + Port int + DefaultDB string + SetBitLogEnabled bool + + Cluster *db.Cluster + ProcessMap *ProcessMap + + Batcher *Batcher + + Executor interface { + RunPQL(database_name string, pql string) (interface{}, error) + } + + Pinger interface { + Ping(process_id *util.GUID) (*time.Duration, error) + } + + TopologyMapper interface { + MakeFragments(db string, slice_int int) error + } + + Transport interface { + Push(message *db.Message) + } } -func NewWebService(service *Service) *WebService { - e := make(chan bool) - return &WebService{service, e} +func NewWebService() *WebService { + return &WebService{ + end: make(chan bool), + } } type Flusher struct { @@ -86,13 +116,13 @@ func NewLogRecord(t time.Time, data []byte) LogRecord { x := LogRecord{t, e} return x } + func genFileName(id string) string { //bucket/YYYY/MM/DDHHMMSS.id.log t := time.Now() //base := "http://pilosa.umbel.com.s3.amazonaws.com/bit_log" - base := config.GetStringDefault("pilosa_request_log", "/tmp/set_bit_log") - return fmt.Sprintf("%s%s.%s.log", base, t.Format("/2006/01/02/15/04-05"), id) + return fmt.Sprintf("%s%s.%s.log", RequestLogPath, t.Format("/2006/01/02/15/04-05"), id) } func flush(requests []LogRecord, id string, records_to_dump int) { @@ -142,7 +172,7 @@ func Logger(in chan []byte, end chan bool, id string, flusher chan bool) { } func (self *WebService) Run() { - port_string := strconv.Itoa(config.GetInt("port_http")) + port_string := strconv.Itoa(self.Port) log.Info("Serving HTTP on port:", port_string) logger_chan := make(chan []byte, 1024) flusher := make(chan bool) @@ -161,8 +191,7 @@ func (self *WebService) Run() { mux.HandleFunc("/ping", self.HandlePing) mux.HandleFunc("/batch", self.HandleBatch) mux.HandleFunc("/load", self.HandleLoad) - log_set_bit := config.GetIntDefault("log_set_bit_request", 0) - if log_set_bit == 1 { + if self.SetBitLogEnabled { mux.HandleFunc("/set_bits", NewRequestLogger(self.HandleSetBit, logger_chan)) } else { mux.HandleFunc("/set_bits", self.HandleSetBit) @@ -174,8 +203,7 @@ func (self *WebService) Run() { Addr: ":" + port_string, Handler: mux, } - id := config.GetString("id") - go Logger(logger_chan, self.end, id, flusher) + go Logger(logger_chan, self.end, self.ID.String(), flusher) s.ListenAndServe() } func (self *WebService) Shutdown() { @@ -194,7 +222,6 @@ func (self *WebService) HandleMessage(w http.ResponseWriter, r *http.Request) { http.Error(w, "Invalid JSON", http.StatusBadRequest) return } - //service.Inbox <- &message } func (self *WebService) HandleLoad(w http.ResponseWriter, r *http.Request) { @@ -221,12 +248,12 @@ func (self *WebService) HandleLoad(w http.ResponseWriter, r *http.Request) { ms_, ok := obj["max_slice"] if ok { ms := int(ms_.(float64)) - database := self.service.Cluster.GetOrCreateDatabase(db) + database := self.Cluster.GetOrCreateDatabase(db) ns, _ := database.NumSlices() if ns <= ms { for i := ns; i <= ms; i++ { log.Info("Load Create Slice ", i) - self.service.TopologyMapper.MakeFragments(db, i) + self.TopologyMapper.MakeFragments(db, i) } http.Error(w, "Needed Slices", http.StatusNotFound) return @@ -261,7 +288,7 @@ func (self *WebService) HandleLoad(w http.ResponseWriter, r *http.Request) { t = float64(obj["filter"].(float64)) filter := uint64(t) - results := FromApiString(self.service, database_name.(string), frame.(string), api_string.(string), bitmap_id, filter) + results := FromApiString(self.Batcher, database_name.(string), frame.(string), api_string.(string), bitmap_id, filter) encoder := json.NewEncoder(w) err = encoder.Encode(results) @@ -323,7 +350,7 @@ func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) { return } - results := self.service.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice), uint64(filter)) + results := self.Batcher.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice), uint64(filter)) encoder := json.NewEncoder(w) err = encoder.Encode(results) @@ -349,13 +376,13 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { database_name := r.Form.Get("db") if database_name == "" { - database_name = config.GetString("default_db") + database_name = self.DefaultDB } if database_name == "" { http.Error(w, "Provide a database (db)", http.StatusNotFound) return } - if !self.service.Cluster.IsValidDatabase(database_name) { + if !self.Cluster.IsValidDatabase(database_name) { http.Error(w, "Unknown Database:"+database_name, http.StatusNotFound) return } @@ -367,7 +394,7 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { _, bits := r.Form["bits"] log.Debug("PQL:", database_name, pql) - results, err := self.service.Executor.RunPQL(database_name, pql) + results, err := self.Executor.RunPQL(database_name, pql) if err != nil { log.Warn("PQL Exec Error:", err.Error(), database_name, pql) http.Error(w, "Error encoding: "+err.Error(), http.StatusInternalServerError) @@ -508,7 +535,13 @@ func (self *WebService) HandleBit(w http.ResponseWriter, r *http.Request, ToSet http.Error(w, "Request To large", http.StatusBadRequest) return } - //remoteSetBit := NewRemoteSetBit(self.service) + + //remoteSetBit := NewRemoteSetBit() + //remoteSetBit.ID = self.ID + //remoteSetBit.ProcessMap = self.ProcessMap + //remoteSetBit.Hold = self.Hold + //remoteSetBit.Transport = self.Transport + for _, obj := range args { if obj["profile_id"] == nil { http.Error(w, "Missing Profile", http.StatusBadRequest) @@ -544,7 +577,7 @@ func (self *WebService) HandleBit(w http.ResponseWriter, r *http.Request, ToSet } else { pql = fmt.Sprintf("clear(%d, %s, %d, %d)", bitmap_id, frame, filter, profile_id) } - result, err := self.service.Executor.RunPQL(dbs, pql) + result, err := self.Executor.RunPQL(dbs, pql) bundle := SBResult{bitmap_id, frame, filter, profile_id, result} results = append(results, bundle) @@ -586,7 +619,7 @@ func (self *WebService) HandleInfo(w http.ResponseWriter, r *http.Request) { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) return } - spew.Fdump(w, self.service.Cluster) + spew.Fdump(w, self.Cluster) } func (self *WebService) HandleVersion(w http.ResponseWriter, r *http.Request) { @@ -595,7 +628,7 @@ func (self *WebService) HandleVersion(w http.ResponseWriter, r *http.Request) { return } - fmt.Fprintf(w, "Pilosa v.("+self.service.version+")\n") + fmt.Fprintf(w, "Pilosa v.("+self.Version+")\n") } func (self *WebService) HandleTest(w http.ResponseWriter, r *http.Request) { @@ -607,11 +640,11 @@ func (self *WebService) HandleTest(w http.ResponseWriter, r *http.Request) { msg := new(db.Message) msg.Data = "mystring" - self.service.Transport.Push(msg) + self.Transport.Push(msg) msg2 := new(db.Message) msg2.Data = 789 - self.service.Transport.Push(msg2) + self.Transport.Push(msg2) } func (self *WebService) HandleProcesses(w http.ResponseWriter, r *http.Request) { @@ -620,7 +653,7 @@ func (self *WebService) HandleProcesses(w http.ResponseWriter, r *http.Request) return } encoder := json.NewEncoder(w) - processes := self.service.ProcessMap.GetMetadata() + processes := self.ProcessMap.GetMetadata() err := encoder.Encode(processes) if err != nil { http.Error(w, "Error Encoding", http.StatusBadRequest) @@ -643,12 +676,12 @@ func (self *WebService) HandlePing(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } - _, err = self.service.ProcessMap.GetProcess(&process_id) + _, err = self.ProcessMap.GetProcess(&process_id) if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return } - duration, err := self.service.Ping(&process_id) + duration, err := self.Pinger.Ping(&process_id) if err != nil { spew.Fdump(w, err) return diff --git a/core/load.go b/core/load.go index c0bd4258c..de07dcdf2 100644 --- a/core/load.go +++ b/core/load.go @@ -20,18 +20,18 @@ func copy_raw(src [32]uint64) index.BlockArray { } return index.BlockArray{o} } -func sendBitmap(service *Service, bitmap index.IBitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) { +func sendBitmap(batcher *Batcher, bitmap index.IBitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) { if slice < 0 { log.Warn("Bad split", db, frame, slice, bitmap_id) finish <- errors.New("BadSplit") return } compressed_bitmap := bitmap.ToCompressString() - results := service.Batch(db, frame, compressed_bitmap, bitmap_id, slice, filter) + results := batcher.Batch(db, frame, compressed_bitmap, bitmap_id, slice, filter) finish <- results } -func FromApiString(service *Service, db string, frame string, api_string string, bitmap_id, filter uint64) string { +func FromApiString(batcher *Batcher, db string, frame string, api_string string, bitmap_id, filter uint64) string { compressed_data, err := base64.StdEncoding.DecodeString(api_string) if err != nil { log.Warn(err) @@ -67,7 +67,7 @@ func FromApiString(service *Service, db string, frame string, api_string string, } else { //make async later sent_count += 1 - go sendBitmap(service, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish) + go sendBitmap(batcher, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish) bitmap = index.NewBitmap() } last_slice = slice @@ -78,7 +78,7 @@ func FromApiString(service *Service, db string, frame string, api_string string, } sent_count += 1 - go sendBitmap(service, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish) + go sendBitmap(batcher, bitmap, db, frame, bitmap_id, filter, int(last_slice), finish) for i := 0; i < sent_count; i++ { <-finish } diff --git a/core/ping.go b/core/ping.go index 644788072..dff2137f9 100644 --- a/core/ping.go +++ b/core/ping.go @@ -29,9 +29,27 @@ func init() { gob.Register(PongRequest{}) } -func (self *Service) Ping(process_id *util.GUID) (*time.Duration, error) { +type Pinger struct { + ID util.GUID + + Hold interface { + Get(id *util.GUID, timeout int) (interface{}, error) + } + + Transport interface { + Send(message *db.Message, host *util.GUID) + } +} + +func NewPinger(id util.GUID) *Pinger { + return &Pinger{ + ID: id, + } +} + +func (self *Pinger) Ping(process_id *util.GUID) (*time.Duration, error) { id := util.RandomUUID() - ping := db.Message{Data: PingRequest{Id: &id, Source: self.Id}} + ping := db.Message{Data: PingRequest{Id: &id, Source: &self.ID}} start := time.Now() self.Transport.Send(&ping, process_id) _, err := self.Hold.Get(&id, 60) diff --git a/core/query.go b/core/query.go deleted file mode 100644 index 867b9ae22..000000000 --- a/core/query.go +++ /dev/null @@ -1,353 +0,0 @@ -package core - -import ( - "sort" - - log "github.com/cihub/seelog" - "github.com/davecgh/go-spew/spew" - "github.com/umbel/pilosa/db" - "github.com/umbel/pilosa/index" - "github.com/umbel/pilosa/query" - "github.com/umbel/pilosa/util" -) - -func (self *Service) CountQueryStepHandler(msg *db.Message) { - log.Trace("CountQueryStepHandler") - qs := msg.Data.(query.CountQueryStep) - input := qs.Input - value, _ := self.Hold.Get(input, util.TimeOut) - var bh index.BitmapHandle - switch val := value.(type) { - case index.BitmapHandle: - bh = val - case []byte: - bh, _ = self.Index.FromBytes(qs.Location.FragmentId, val) - } - count, err := self.Index.Count(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - result_message := db.Message{Data: query.CountQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: count}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) UnionQueryStepHandler(msg *db.Message) { - log.Trace("UnionQueryStepHandler") - qs := msg.Data.(query.UnionQueryStep) - var handles []index.BitmapHandle - // create a list of bitmap handles - for _, input := range qs.Inputs { - value, _ := self.Hold.Get(input, util.TimeOut) - switch val := value.(type) { - case index.BitmapHandle: - handles = append(handles, val) - case []byte: - bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) - handles = append(handles, bh) - } - } - - bh, err := self.Index.Union(qs.Location.FragmentId, handles) - if err != nil { - spew.Dump(err) - } - - var result interface{} - if qs.LocIsDest() { - result = bh - } else { - bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - result = bm - } - result_message := db.Message{Data: query.UnionQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) IntersectQueryStepHandler(msg *db.Message) { - log.Trace("IntersectQueryStepHandler") - qs := msg.Data.(query.IntersectQueryStep) - var handles []index.BitmapHandle - // create a list of bitmap handles - for _, input := range qs.Inputs { - value, _ := self.Hold.Get(input, util.TimeOut) - switch val := value.(type) { - case index.BitmapHandle: - handles = append(handles, val) - case []byte: - bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) - handles = append(handles, bh) - } - } - - bh, err := self.Index.Intersect(qs.Location.FragmentId, handles) - if err != nil { - spew.Dump(err) - } - - var result interface{} - if qs.LocIsDest() { - result = bh - } else { - bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - result = bm - } - result_message := db.Message{Data: query.IntersectQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) DifferenceQueryStepHandler(msg *db.Message) { - log.Trace("DifferenceQueryStepHandler") - qs := msg.Data.(query.DifferenceQueryStep) - var handles []index.BitmapHandle - // create a list of bitmap handles - for _, input := range qs.Inputs { - value, _ := self.Hold.Get(input, util.TimeOut) - switch val := value.(type) { - case index.BitmapHandle: - handles = append(handles, val) - case []byte: - bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) - handles = append(handles, bh) - } - } - - bh, err := self.Index.Difference(qs.Location.FragmentId, handles) - if err != nil { - spew.Dump(err) - } - - var result interface{} - if qs.LocIsDest() { - result = bh - } else { - bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - result = bm - } - result_message := db.Message{Data: query.DifferenceQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) StashQueryStepHandler(msg *db.Message) { - log.Trace("StashQueryStepHandler") - qs := msg.Data.(query.StashQueryStep) - - part := make(chan interface{}) - num_parts := len(qs.Inputs) - - for _, input := range qs.Inputs { - go func(id *util.GUID, part chan interface{}) { - value, _ := self.Hold.Get(id, util.TimeOut) - part <- value - }(input, part) - } - //just collect all the handles and return them - result := query.NewStash() //query.Stash{make([]query.CacheItem, 0), false} - for i := 0; i < num_parts; i++ { - value := <-part - - switch val := value.(type) { - case index.BitmapHandle: - log.Info("STASH ADDING HANDLE", val) - //not sure what to do here.... - //result.Handles = append(result.Handles, val) - case []byte: - bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) - item := query.CacheItem{qs.Location.FragmentId, bh} - result.Stash = append(result.Stash, item) - case query.Stash: - result.Stash = append(result.Stash, val.Stash...) - default: - log.Warn("UNEXCPECTED MESSAGE", value) - } - } - result_message := db.Message{Data: query.StashQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) - -} - -func (self *Service) CatQueryStepHandler(msg *db.Message) { - log.Trace("CatQueryStepHandler") - qs := msg.Data.(query.CatQueryStep) - var handles []index.BitmapHandle - return_type := "bitmap-handles" - var sum uint64 - merge_map := make(map[uint64]uint64) - slice_map := make(map[uint64]map[util.SUUID]struct{}) - all_slice := make(map[util.SUUID]struct { - process util.GUID - handle index.BitmapHandle - }) - - // either create a list of bitmap handles to cat (i.e. union), or sum the integer values - part := make(chan interface{}) - num_parts := len(qs.Inputs) - - for _, input := range qs.Inputs { - go func(id *util.GUID, part chan interface{}) { - value, _ := self.Hold.Get(id, util.TimeOut) - part <- value - }(input, part) - } - - //for _, input := range qs.Inputs { - check_pair := false - for i := 0; i < num_parts; i++ { - value := <-part - - switch val := value.(type) { - case index.BitmapHandle: - handles = append(handles, val) - case []byte: - bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) - handles = append(handles, bh) - case uint64: - //spew.Dump(val) - return_type = "sum" - sum += val - case TopNPackage: - return_type = "pair-list" - var e struct{} - for _, pair := range val.Pairs { - //merge_map[pair.Key] += pair.Count - if pair.Key == 0 { - continue //skip - } - merge_map[pair.Key] += pair.Count - mm, ok := slice_map[pair.Key] - if !ok { - mm = make(map[util.SUUID]struct{}) - slice_map[pair.Key] = mm - } - mm[val.FragmentId] = e - } - all_slice[val.FragmentId] = struct { - process util.GUID - handle index.BitmapHandle - }{val.ProcessId, val.HBitmap} - check_pair = true - } - } - if check_pair { //no point in doing this for non top-n handling - tasks := BuildTask(merge_map, slice_map, all_slice) - FetchMissing(tasks, self) - for k, v := range GatherResults(tasks, self) { - merge_map[k] += v - } - } - - // either return the sum, or return the compressed bitmap resulting from the cat (union) - var result interface{} - if return_type == "sum" { - result = sum - } else if return_type == "bitmap-handles" { - bh, err := self.Index.Union(qs.Location.FragmentId, handles) - result, err = self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - } else if return_type == "pair-list" { - rank_list := make(index.RankList, 0, len(merge_map)) - for k, v := range merge_map { - if k == 0 || v == 0 { - continue //shouldn't be getting 0 keys or values anyway - } - rank := new(index.Rank) - rank.Pair = &index.Pair{k, v} - rank_list = append(rank_list, rank) - } - sort.Sort(rank_list) // kinda seems like this copy is wasteful..i'll ponder - items_size := min(len(merge_map), qs.N) - pair_list := make([]index.Pair, 0, items_size+1) - for i, r := range rank_list { - if i < items_size { - pair_list = append(pair_list, *r.Pair) - } else { - break - } - } - result = pair_list - } else { - result = "NONE" - } - result_message := db.Message{Data: query.CatQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func min(a, b int) int { - if a < b { - return a - } - return b -} -func (self *Service) GetQueryStepHandler(msg *db.Message) { - qs := msg.Data.(query.GetQueryStep) - - bh, err := self.Index.Get(qs.Location.FragmentId, qs.Bitmap.Id) - if err != nil { - spew.Dump(err) - log.Error("GetQueryStepHandler1", util.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) - log.Error("GetQueryStepHandler2", err) - } - - var result interface{} - if qs.LocIsDest() { - result = bh - } else { - bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - log.Error("GetQueryStepHandlerr3", util.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) - log.Error("GetQueryStepHandler4", err) - } - result = bm - } - result_message := db.Message{Data: query.GetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) SetQueryStepHandler(msg *db.Message) { - qs := msg.Data.(query.SetQueryStep) - result, _ := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId, qs.Bitmap.Filter) - - result_message := db.Message{Data: query.SetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) ClearQueryStepHandler(msg *db.Message) { - qs := msg.Data.(query.ClearQueryStep) - result, _ := self.Index.ClearBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) - - result_message := db.Message{Data: query.ClearQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func (self *Service) RangeQueryStepHandler(msg *db.Message) { - qs := msg.Data.(query.RangeQueryStep) - - bh, err := self.Index.Range(qs.Location.FragmentId, qs.Bitmap.Id, qs.Start, qs.End) - if err != nil { - spew.Dump(err) - } - - var result interface{} - if qs.LocIsDest() { - result = bh - } else { - bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - result = bm - } - result_message := db.Message{Data: query.RangeQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} diff --git a/core/remotebits.go b/core/remotebits.go index 70e7be0fa..3b489d171 100644 --- a/core/remotebits.go +++ b/core/remotebits.go @@ -11,7 +11,17 @@ import ( type RemoteSetBit struct { requests []remote_task cluster map[*util.GUID][]BitmapRequestItem - service *Service + + ID util.GUID + ProcessMap *ProcessMap + + Hold interface { + Get(id *util.GUID, timeout int) (interface{}, error) + } + + Transport interface { + Send(message *db.Message, host *util.GUID) + } } func init() { @@ -35,16 +45,15 @@ type BitmapRequestItem struct { SetUnset bool } -func NewRemoteSetBit(s *Service) *RemoteSetBit { +func NewRemoteSetBit() *RemoteSetBit { obj := new(RemoteSetBit) obj.cluster = make(map[*util.GUID][]BitmapRequestItem) - obj.service = s return obj } func (self *RemoteSetBit) Request() { self.requests = make([]remote_task, 0) - source_process, _ := self.service.GetProcess() + source_process, _ := self.ProcessMap.GetProcess(&self.ID) for process, request := range self.cluster { random_id := util.RandomUUID() msg := new(db.Message) @@ -59,7 +68,7 @@ func (self *RemoteSetBit) Request() { wait = 10 } self.requests = append(self.requests, remote_task{random_id, wait}) - self.service.Transport.Send(msg, process) + self.Transport.Send(msg, process) } } @@ -73,7 +82,7 @@ func (self *RemoteSetBit) MergeResults(local_results []SBResult) []SBResult { answers := make(chan []SBResult) for _, task := range self.requests { go func(task remote_task) { - value, err := self.service.Hold.Get(&task.id, task.wait_time) //eiher need to be the frame process or the handler process? + value, err := self.Hold.Get(&task.id, task.wait_time) //eiher need to be the frame process or the handler process? if value == nil { log.Warn("Bad RemoteSetBit Result:", err) empty := make([]SBResult, 0, 0) diff --git a/core/service.go b/core/service.go index 2efcc3010..4c888624b 100644 --- a/core/service.go +++ b/core/service.go @@ -1,14 +1,12 @@ package core import ( - "fmt" "os" "os/signal" "syscall" log "github.com/cihub/seelog" "github.com/coreos/go-etcd/etcd" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/db" "github.com/umbel/pilosa/hold" "github.com/umbel/pilosa/index" @@ -24,8 +22,6 @@ type Service struct { TopologyMapper *TopologyMapper ProcessMapper *ProcessMapper ProcessMap *ProcessMap - Transport interfaces.Transporter - Dispatch interfaces.Dispatcher Executor interfaces.Executorer WebService *WebService Index *index.FragmentContainer @@ -36,85 +32,6 @@ type Service struct { var Build string -func NewService() *Service { - service := new(Service) - service.init_id() - etc_hosts := config.GetStringArrayDefault("etcd_servers", []string{}) - - service.Etcd = etcd.NewClient(etc_hosts) - service.Cluster = db.NewCluster() - service.TopologyMapper = NewTopologyMapper(service, "/pilosa/0") - service.ProcessMapper = NewProcessMapper(service, "/pilosa/0") - service.ProcessMap = NewProcessMap() - service.WebService = NewWebService(service) - service.Index = index.NewFragmentContainer() - service.Hold = hold.NewHolder() - service.version = Build - service.name = "Cruncher" - service.PrepareLogging() - fmt.Printf("Pilosa %s\n", service.version) - return service -} - -func (self *Service) getProduction() string { - base_path := config.GetString("log_path") - if base_path == "" { - base_path = "/tmp" - } - fname := fmt.Sprintf("%s/%s.%s", base_path, self.name, self.Id) - // - log_level := config.GetStringDefault("log_level", "info") - prod_config := fmt.Sprintf(` - - - `, log_level, fname) - prod_config += ` - - - ` - //fmt.Println(prod_config) - return prod_config -} - -func (self *Service) getDev() string { - return ` - - - - - ` -} -func (self *Service) PrepareLogging() { - logger, _ := log.LoggerFromConfigAsBytes([]byte(self.getProduction())) - log.ReplaceLogger(logger) -} - -func (service *Service) init_id() { - var id util.GUID - var err error - id_string := config.GetString("id") - if id_string == "" { - log.Info("Service id not configured, generating...") - id = util.RandomUUID() - if err != nil { - log.Critical("problem generating uuid") - os.Exit(-1) - - } - } else { - id, err = util.ParseGUID(id_string) - if err != nil { - log.Critical("Service id not valid:", id_string) - os.Exit(-1) - } - } - service.Id = &id -} - -func (self *Service) GetProcess() (*db.Process, error) { - return self.ProcessMap.GetProcess(self.Id) -} - func (service *Service) GetSignals() (chan os.Signal, chan os.Signal) { hupChan := make(chan os.Signal, 1) termChan := make(chan os.Signal, 1) @@ -130,8 +47,6 @@ func (service *Service) Run() { go service.TopologyMapper.Run() go service.ProcessMapper.Run() go service.WebService.Run() - go service.Transport.Run() - go service.Dispatch.Run() go service.Executor.Run() go service.Hold.Run() diff --git a/core/topn.go b/core/topn.go deleted file mode 100644 index c3843437b..000000000 --- a/core/topn.go +++ /dev/null @@ -1,199 +0,0 @@ -package core - -import ( - "encoding/gob" - - log "github.com/cihub/seelog" - "github.com/davecgh/go-spew/spew" - "github.com/umbel/pilosa/db" - "github.com/umbel/pilosa/index" - "github.com/umbel/pilosa/query" - "github.com/umbel/pilosa/util" -) - -type Task struct { - processid util.GUID - f map[util.SUUID]index.FillArgs - hold_id util.GUID -} - -type TopFill struct { - Args []index.FillArgs - ReturnProcessId util.GUID - QueryId util.GUID - DestProcessId util.GUID -} - -//portable query step -func (self *TopFill) GetId() *util.GUID { - return &self.QueryId -} -func (self *TopFill) GetLocation() *db.Location { - return &db.Location{&self.DestProcessId, 0} //this message is a broadcast to many fragments so i'm choosing fragmentzero -} - -// - -type hole struct { - process util.GUID - handle index.BitmapHandle - fragment util.SUUID -} - -func missing(fids map[util.SUUID]struct{}, all map[util.SUUID]struct { - process util.GUID - handle index.BitmapHandle -}) []hole { - results := make([]hole, 0, 0) - - for k, v := range all { - _, ok := fids[k] - if !ok { - results = append(results, hole{v.process, v.handle, k}) - } - } - return results -} - -func newtask(p util.GUID) *Task { - result := new(Task) - result.processid = p - result.f = make(map[util.SUUID]index.FillArgs) - result.hold_id = util.RandomUUID() - return result -} - -func (t *Task) Add(frag util.SUUID, bitmap_id uint64, handle index.BitmapHandle) { - fa, ok := t.f[frag] - if !ok { - fa = index.FillArgs{frag, handle, make([]uint64, 0, 0)} - } - fa.Bitmaps = append(fa.Bitmaps, bitmap_id) - t.f[frag] = fa -} - -func BuildTask(merge_map map[uint64]uint64, - slice_map map[uint64]map[util.SUUID]struct{}, - total_fragments map[util.SUUID]struct { - process util.GUID - handle index.BitmapHandle - }) map[util.GUID]*Task { - - tasks := make(map[util.GUID]*Task) - for bitmap_id, _ := range merge_map { //for all brands - reporting_fragments := slice_map[bitmap_id] - for _, p := range missing(reporting_fragments, total_fragments) { - task, ok := tasks[p.process] - if !ok { - task = newtask(p.process) - tasks[p.process] = task - } - task.Add(p.fragment, bitmap_id, p.handle) - - } - } - - return tasks -} -func (self *Service) TopFillHandler(msg *db.Message) { //in order for this to get executed it needs to be a portable query step - topfill := msg.Data.(TopFill) - topn, err := self.Index.TopFillBatch(topfill.Args) - if err != nil { - log.Warn("TopFillHandler:", err) - } - - result_message := db.Message{Data: query.FillResult{&query.BaseQueryResult{Id: &topfill.QueryId, Data: topn}}} - self.Transport.Send(&result_message, &topfill.ReturnProcessId) -} - -func SendRequest(process_id util.GUID, t *Task, service *Service) { - args := make([]index.FillArgs, len(t.f), len(t.f)) - for _, v := range t.f { - args = append(args, v) - } - msg := new(db.Message) - p, _ := service.GetProcess() - msg.Data = TopFill{args, p.Id(), t.hold_id, process_id} - service.Transport.Send(msg, &process_id) -} - -func FetchMissing(tasks map[util.GUID]*Task, service *Service) { - for k, v := range tasks { - go SendRequest(k, v, service) - } -} - -func GatherResults(tasks map[util.GUID]*Task, service *Service) map[uint64]uint64 { - results := make(map[uint64]uint64) - answers := make(chan []index.Pair) - for _, task := range tasks { - go func(id util.GUID) { - value, err := service.Hold.Get(&id, 10) //eiher need to be the frame process or the handler process? - if value == nil { - log.Warn("Bad TopN Result:", err) - empty := make([]index.Pair, 0, 0) - answers <- empty - - } else { - answers <- value.([]index.Pair) - } - }(task.hold_id) - } - for i := 0; i < len(tasks); i++ { - batch := <-answers - for _, pair := range batch { - results[pair.Key] += pair.Count - } - } - close(answers) - return results -} - -type TopNPackage struct { - ProcessId util.GUID - FragmentId util.SUUID - Pairs []index.Pair - HBitmap index.BitmapHandle -} - -func init() { - gob.Register(TopNPackage{}) - gob.Register(TopFill{}) -} - -func (self *Service) TopNQueryStepHandler(msg *db.Message) { - qs := msg.Data.(query.TopNQueryStep) - var bh index.BitmapHandle - var topnPackage TopNPackage - - // if we have an input, hold for it. if we don't, we assume an all() query - if qs.Input == nil { - topn, err := self.Index.TopNAll(qs.Location.FragmentId, qs.N*2, qs.Filters) - if err != nil { - log.Warn(spew.Sdump(err)) - } - topnPackage = TopNPackage{*qs.Location.ProcessId, qs.Location.FragmentId, topn, bh} - } else { - input := qs.Input - value, _ := self.Hold.Get(input, 10) - switch val := value.(type) { - case index.BitmapHandle: - bh = val - case []byte: - bh, _ = self.Index.FromBytes(qs.Location.FragmentId, val) - } - - topn, err := self.Index.TopN(qs.Location.FragmentId, bh, qs.N*2, qs.Filters) - if err != nil { - log.Warn(spew.Sdump(err)) - } - topnPackage = TopNPackage{*qs.Location.ProcessId, qs.Location.FragmentId, topn, bh} - } - - result_message := db.Message{Data: query.TopNQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: topnPackage}}} - self.Transport.Send(&result_message, qs.Destination.ProcessId) -} - -func canCompile() bool { - return true -} diff --git a/db/topology.go b/db/topology.go index 2075c1229..adda65714 100644 --- a/db/topology.go +++ b/db/topology.go @@ -7,10 +7,12 @@ import ( log "github.com/cihub/seelog" "github.com/stathat/consistent" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/util" ) +// SupportedFrames is a list of frame types that are supported. +var SupportedFrames = []string{"default"} + var FrameDoesNotExistError = errors.New("Frame does not exist.") var InvalidFrameError = errors.New("Invalid frame.") var SliceDoesNotExistError = errors.New("Slice does not exist.") @@ -161,9 +163,7 @@ func stringInSlice(a string, list []string) bool { } func (d *Database) IsValidFrame(name string) bool { - supported_frames := config.GetStringArrayDefault("supported_frames", - []string{"default"}) - return stringInSlice(name, supported_frames) + return stringInSlice(name, SupportedFrames) } // Count the number of slices in a database diff --git a/default.yaml b/default.yaml index 640910d92..2e2c8dc05 100644 --- a/default.yaml +++ b/default.yaml @@ -14,4 +14,4 @@ supported_frames: - d - p.n etcd_servers: - - http://127.0.0.1:4001 + - http://127.0.0.1:4001 diff --git a/deps.json b/deps.json index b61836739..0d3df27da 100644 --- a/deps.json +++ b/deps.json @@ -94,6 +94,11 @@ "version": "3999011ef0451eb805b0fa31f98cfb9261bd71be", "type": "git" }, + "toml": { + "repo": "github.com/BurntSushi/toml", + "version": "056c9bc7be7190eaa7715723883caffa5f8fa3e4", + "type": "git" + }, "websocket": { "repo": "github.com/gorilla/websocket", "version": "92334662baa9cbebc2e6e68b8d56bc1233f85a4c", diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 84e1043d4..c1eb56258 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -5,13 +5,40 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/umbel/pilosa/core" "github.com/umbel/pilosa/db" + "github.com/umbel/pilosa/executor" + "github.com/umbel/pilosa/index" "github.com/umbel/pilosa/query" + "github.com/umbel/pilosa/util" ) type Dispatch struct { + Executor interface { + NewJob(job *db.Message) + } + + Hold interface { + Set(id *util.GUID, value interface{}, timeout int) + } + + Index interface { + ClearBit(fragID util.SUUID, bitmapID uint64, pos uint64) (bool, error) + LoadBitmap(fragID util.SUUID, bitmapID uint64, compressedBitmap string, filter uint64) + SetBit(fragID util.SUUID, bitmapID uint64, pos uint64, category uint64) (bool, error) + TopFillBatch(args []index.FillArgs) ([]index.Pair, error) + } + + Transport interface { + Receive() *db.Message + Send(message *db.Message, host *util.GUID) + } + service *core.Service } +func NewDispatch() *Dispatch { + return &Dispatch{} +} + func (self *Dispatch) Init() error { log.Warn("Starting Dispatcher") return nil @@ -23,16 +50,18 @@ func (self *Dispatch) Close() { // The Local Route -func (self *Dispatch) Run() { +func (d *Dispatch) Run() { log.Warn("Dispatch Run...") for { - message := self.service.Transport.Receive() + message := d.Transport.Receive() + switch data := message.Data.(type) { case core.BatchRequest: log.Trace("Dispatch.Run BatchRequest") response := db.Message{Data: core.BatchResponse{Id: data.Id}} - self.service.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap, data.Filter) - self.service.Transport.Send(&response, data.Source) + d.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap, data.Filter) + d.Transport.Send(&response, data.Source) + case core.BitsRequest: log.Trace("Dispatch.Run BitsRequest") var results []core.SBResult @@ -40,30 +69,36 @@ func (self *Dispatch) Run() { for _, v := range data.Bits { if v.SetUnset { - result, _ = self.service.Index.SetBit(v.Fragment_id, v.Bitmap_id, v.Profile_id, uint64(v.Filter)) + result, _ = d.Index.SetBit(v.Fragment_id, v.Bitmap_id, v.Profile_id, uint64(v.Filter)) } else { - result, _ = self.service.Index.ClearBit(v.Fragment_id, v.Bitmap_id, v.Profile_id) + result, _ = d.Index.ClearBit(v.Fragment_id, v.Bitmap_id, v.Profile_id) } bundle := core.SBResult{v.Bitmap_id, v.Frame, v.Filter, v.Profile_id, result} results = append(results, bundle) } response := db.Message{Data: core.BitsResponse{Id: &data.QueryId, Items: results}} - self.service.Transport.Send(&response, &data.ReturnProcessId) + d.Transport.Send(&response, &data.ReturnProcessId) + case core.PingRequest: log.Trace("Dispatch.Run Ping") pong := db.Message{Data: core.PongRequest{Id: data.Id}} - self.service.Transport.Send(&pong, data.Source) + d.Transport.Send(&pong, data.Source) + case db.HoldResult: log.Trace("Dispatch.Run HoldResult") - self.service.Hold.Set(data.ResultId(), data.ResultData(), 30) + d.Hold.Set(data.ResultId(), data.ResultData(), 30) + case query.PortableQueryStep: log.Trace("Dispatch.Run PortableQueryStep") - go self.service.Executor.NewJob(message) - case core.TopFill: + go d.Executor.NewJob(message) + + case executor.TopFill: log.Trace("Dispatch.Run TopFill") - go self.service.TopFillHandler(message) + go d.topFillHandler(message) + case core.BitsResponse: - self.service.Hold.Set(data.ResultId(), data.ResultData(), 30) + d.Hold.Set(data.ResultId(), data.ResultData(), 30) + default: spew.Dump(data) log.Warn("Unprocessed message", data) @@ -71,6 +106,19 @@ func (self *Dispatch) Run() { } } -func NewDispatch(service *core.Service) *Dispatch { - return &Dispatch{service} +func (d *Dispatch) topFillHandler(m *db.Message) { + topfill := m.Data.(executor.TopFill) + topn, err := d.Index.TopFillBatch(topfill.Args) + if err != nil { + log.Warn("TopFillHandler:", err) + } + + d.Transport.Send(&db.Message{ + Data: query.FillResult{ + &query.BaseQueryResult{ + Id: &topfill.QueryId, + Data: topn, + }, + }, + }, &topfill.ReturnProcessId) } diff --git a/executor/executor.go b/executor/executor.go index 8fbeae279..0fbbb4603 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -1,18 +1,59 @@ package executor import ( + "encoding/gob" + "sort" + "time" + log "github.com/cihub/seelog" "github.com/davecgh/go-spew/spew" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/core" "github.com/umbel/pilosa/db" + "github.com/umbel/pilosa/index" "github.com/umbel/pilosa/query" "github.com/umbel/pilosa/util" ) type Executor struct { - service *core.Service - inbox chan *db.Message + inbox chan *db.Message + + ID util.GUID + Cluster *db.Cluster + ProcessMap *core.ProcessMap + PluginsPath string + + Hold interface { + Get(id *util.GUID, timeout int) (interface{}, error) + Set(id *util.GUID, value interface{}, timeout int) + } + + Index interface { + ClearBit(frag_id util.SUUID, bitmap_id uint64, pos uint64) (bool, error) + Count(frag_id util.SUUID, bitmap index.BitmapHandle) (uint64, error) + Difference(frag_id util.SUUID, bh []index.BitmapHandle) (index.BitmapHandle, error) + FromBytes(frag_id util.SUUID, bytes []byte) (index.BitmapHandle, error) + Get(frag_id util.SUUID, bitmap_id uint64) (index.BitmapHandle, error) + GetBytes(frag_id util.SUUID, bh index.BitmapHandle) ([]byte, error) + Intersect(frag_id util.SUUID, bh []index.BitmapHandle) (index.BitmapHandle, error) + Range(frag_id util.SUUID, bitmap_id uint64, start, end time.Time) (index.BitmapHandle, error) + SetBit(frag_id util.SUUID, bitmap_id uint64, pos uint64, category uint64) (bool, error) + TopN(frag_id util.SUUID, bh index.BitmapHandle, n int, categories []uint64) ([]index.Pair, error) + TopNAll(frag_id util.SUUID, n int, categories []uint64) ([]index.Pair, error) + Union(frag_id util.SUUID, bh []index.BitmapHandle) (index.BitmapHandle, error) + } + + TopologyMapper interface { + MakeFragments(db string, slice_int int) error + } + + Transport interface { + Send(*db.Message, *util.GUID) + } +} + +func NewExecutor(id util.GUID) *Executor { + log.Trace("NewExector") + return &Executor{inbox: make(chan *db.Message)} } func (self *Executor) Init() error { @@ -28,42 +69,453 @@ func (self *Executor) NewJob(job *db.Message) { log.Trace("NewJob", job) switch job.Data.(type) { case query.CountQueryStep: - self.service.CountQueryStepHandler(job) + self.CountQueryStepHandler(job) case query.TopNQueryStep: - self.service.TopNQueryStepHandler(job) + self.TopNQueryStepHandler(job) case query.UnionQueryStep: - self.service.UnionQueryStepHandler(job) + self.UnionQueryStepHandler(job) case query.IntersectQueryStep: - self.service.IntersectQueryStepHandler(job) + self.IntersectQueryStepHandler(job) case query.DifferenceQueryStep: - self.service.DifferenceQueryStepHandler(job) + self.DifferenceQueryStepHandler(job) case query.CatQueryStep: - self.service.CatQueryStepHandler(job) + self.CatQueryStepHandler(job) case query.GetQueryStep: - self.service.GetQueryStepHandler(job) + self.GetQueryStepHandler(job) case query.SetQueryStep: - self.service.SetQueryStepHandler(job) + self.SetQueryStepHandler(job) case query.ClearQueryStep: - self.service.ClearQueryStepHandler(job) + self.ClearQueryStepHandler(job) case query.RangeQueryStep: - self.service.RangeQueryStepHandler(job) + self.RangeQueryStepHandler(job) case query.StashQueryStep: - self.service.StashQueryStepHandler(job) + self.StashQueryStepHandler(job) default: log.Warn("unknown") log.Warn(spew.Sdump(job.Data)) } } -type stringSlice []string +func (self *Executor) CountQueryStepHandler(msg *db.Message) { + log.Trace("CountQueryStepHandler") + //spew.Dump("COUNT QUERYSTEP") + qs := msg.Data.(query.CountQueryStep) + input := qs.Input + value, _ := self.Hold.Get(input, util.TimeOut) + var bh index.BitmapHandle + switch val := value.(type) { + case index.BitmapHandle: + bh = val + case []byte: + bh, _ = self.Index.FromBytes(qs.Location.FragmentId, val) + } + count, err := self.Index.Count(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + //spew.Dump("SLICE COUNT", count) + result_message := db.Message{Data: query.CountQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: count}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} -func (slice stringSlice) pos(value string) int { - for p, v := range slice { - if v == value { - return p +func (self *Executor) TopNQueryStepHandler(msg *db.Message) { + qs := msg.Data.(query.TopNQueryStep) + var bh index.BitmapHandle + var topnPackage TopNPackage + + // if we have an input, hold for it. if we don't, we assume an all() query + if qs.Input == nil { + topn, err := self.Index.TopNAll(qs.Location.FragmentId, qs.N*2, qs.Filters) + if err != nil { + log.Warn(spew.Sdump(err)) + } + topnPackage = TopNPackage{*qs.Location.ProcessId, qs.Location.FragmentId, topn, bh} + } else { + input := qs.Input + value, _ := self.Hold.Get(input, 10) + //var bh index.BitmapHandle + switch val := value.(type) { + case index.BitmapHandle: + bh = val + case []byte: + bh, _ = self.Index.FromBytes(qs.Location.FragmentId, val) + } + + topn, err := self.Index.TopN(qs.Location.FragmentId, bh, qs.N*2, qs.Filters) + if err != nil { + log.Warn(spew.Sdump(err)) + } + topnPackage = TopNPackage{*qs.Location.ProcessId, qs.Location.FragmentId, topn, bh} + } + + result_message := db.Message{Data: query.TopNQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: topnPackage}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) UnionQueryStepHandler(msg *db.Message) { + log.Trace("UnionQueryStepHandler") + //spew.Dump("UNION QUERYSTEP") + + qs := msg.Data.(query.UnionQueryStep) + var handles []index.BitmapHandle + // create a list of bitmap handles + for _, input := range qs.Inputs { + value, _ := self.Hold.Get(input, util.TimeOut) + switch val := value.(type) { + case index.BitmapHandle: + handles = append(handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + handles = append(handles, bh) } } - return -1 + + bh, err := self.Index.Union(qs.Location.FragmentId, handles) + if err != nil { + spew.Dump(err) + } + + var result interface{} + if qs.LocIsDest() { + result = bh + } else { + bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + result = bm + } + result_message := db.Message{Data: query.UnionQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) IntersectQueryStepHandler(msg *db.Message) { + log.Trace("IntersectQueryStepHandler") + //spew.Dump("INTERSECT QUERYSTEP") + qs := msg.Data.(query.IntersectQueryStep) + var handles []index.BitmapHandle + // create a list of bitmap handles + for _, input := range qs.Inputs { + value, _ := self.Hold.Get(input, util.TimeOut) + switch val := value.(type) { + case index.BitmapHandle: + handles = append(handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + handles = append(handles, bh) + } + } + + bh, err := self.Index.Intersect(qs.Location.FragmentId, handles) + if err != nil { + spew.Dump(err) + } + + var result interface{} + if qs.LocIsDest() { + result = bh + } else { + bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + result = bm + } + result_message := db.Message{Data: query.IntersectQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) DifferenceQueryStepHandler(msg *db.Message) { + log.Trace("DifferenceQueryStepHandler") + //spew.Dump("DIFFERENCE QUERYSTEP") + qs := msg.Data.(query.DifferenceQueryStep) + var handles []index.BitmapHandle + // create a list of bitmap handles + for _, input := range qs.Inputs { + value, _ := self.Hold.Get(input, util.TimeOut) + switch val := value.(type) { + case index.BitmapHandle: + handles = append(handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + handles = append(handles, bh) + } + } + + bh, err := self.Index.Difference(qs.Location.FragmentId, handles) + if err != nil { + spew.Dump(err) + } + + var result interface{} + if qs.LocIsDest() { + result = bh + } else { + bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + result = bm + } + result_message := db.Message{Data: query.DifferenceQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) CatQueryStepHandler(msg *db.Message) { + log.Trace("CatQueryStepHandler") + qs := msg.Data.(query.CatQueryStep) + var handles []index.BitmapHandle + return_type := "bitmap-handles" + var sum uint64 + merge_map := make(map[uint64]uint64) + slice_map := make(map[uint64]map[util.SUUID]struct{}) + all_slice := make(map[util.SUUID]struct { + process util.GUID + handle index.BitmapHandle + }) + + // either create a list of bitmap handles to cat (i.e. union), or sum the integer values + part := make(chan interface{}) + num_parts := len(qs.Inputs) + + for _, input := range qs.Inputs { + go func(id *util.GUID, part chan interface{}) { + value, _ := self.Hold.Get(id, util.TimeOut) + part <- value + }(input, part) + } + + //for _, input := range qs.Inputs { + check_pair := false + for i := 0; i < num_parts; i++ { + value := <-part + + switch val := value.(type) { + case index.BitmapHandle: + handles = append(handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + handles = append(handles, bh) + case uint64: + //spew.Dump(val) + return_type = "sum" + sum += val + case TopNPackage: + return_type = "pair-list" + var e struct{} + for _, pair := range val.Pairs { + //merge_map[pair.Key] += pair.Count + if pair.Key == 0 { + continue //skip + } + merge_map[pair.Key] += pair.Count + mm, ok := slice_map[pair.Key] + if !ok { + mm = make(map[util.SUUID]struct{}) + slice_map[pair.Key] = mm + } + mm[val.FragmentId] = e + } + all_slice[val.FragmentId] = struct { + process util.GUID + handle index.BitmapHandle + }{val.ProcessId, val.HBitmap} + check_pair = true + } + } + if check_pair { //no point in doing this for non top-n handling + tasks := BuildTask(merge_map, slice_map, all_slice) + self.FetchMissing(tasks) + for k, v := range self.GatherResults(tasks) { + merge_map[k] += v + } + } + + // either return the sum, or return the compressed bitmap resulting from the cat (union) + var result interface{} + if return_type == "sum" { + result = sum + } else if return_type == "bitmap-handles" { + bh, err := self.Index.Union(qs.Location.FragmentId, handles) + result, err = self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + } else if return_type == "pair-list" { + rank_list := make(index.RankList, 0, len(merge_map)) + for k, v := range merge_map { + if k == 0 || v == 0 { + continue //shouldn't be getting 0 keys or values anyway + } + rank := new(index.Rank) + rank.Pair = &index.Pair{k, v} + rank_list = append(rank_list, rank) + } + sort.Sort(rank_list) // kinda seems like this copy is wasteful..i'll ponder + items_size := min(len(merge_map), qs.N) + pair_list := make([]index.Pair, 0, items_size+1) + for i, r := range rank_list { + if i < items_size { + pair_list = append(pair_list, *r.Pair) + } else { + break + } + } + result = pair_list + } else { + result = "NONE" + } + result_message := db.Message{Data: query.CatQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) SendRequest(process_id util.GUID, t *Task) { + args := make([]index.FillArgs, len(t.f), len(t.f)) + for _, v := range t.f { + args = append(args, v) + } + msg := new(db.Message) + p, _ := self.ProcessMap.GetProcess(&self.ID) + msg.Data = TopFill{args, p.Id(), t.hold_id, process_id} + self.Transport.Send(msg, &process_id) +} + +func (self *Executor) FetchMissing(tasks map[util.GUID]*Task) { + for k, v := range tasks { + go self.SendRequest(k, v) + } +} + +func (self *Executor) GatherResults(tasks map[util.GUID]*Task) map[uint64]uint64 { + results := make(map[uint64]uint64) + answers := make(chan []index.Pair) + for _, task := range tasks { + go func(id util.GUID) { + value, err := self.Hold.Get(&id, 10) //eiher need to be the frame process or the handler process? + if value == nil { + log.Warn("Bad TopN Result:", err) + empty := make([]index.Pair, 0, 0) + answers <- empty + + } else { + answers <- value.([]index.Pair) + } + }(task.hold_id) + } + for i := 0; i < len(tasks); i++ { + batch := <-answers + for _, pair := range batch { + results[pair.Key] += pair.Count + } + } + close(answers) + return results +} + +func (self *Executor) GetQueryStepHandler(msg *db.Message) { + qs := msg.Data.(query.GetQueryStep) + //spew.Dump("GET QUERYSTEP") + + bh, err := self.Index.Get(qs.Location.FragmentId, qs.Bitmap.Id) + if err != nil { + spew.Dump(err) + log.Error("GetQueryStepHandler1", util.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) + log.Error("GetQueryStepHandler2", err) + } + + var result interface{} + if qs.LocIsDest() { + result = bh + } else { + bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + log.Error("GetQueryStepHandlerr3", util.SUUID_to_Hex(qs.Location.FragmentId), qs.Bitmap.Id) + log.Error("GetQueryStepHandler4", err) + } + result = bm + } + result_message := db.Message{Data: query.GetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) SetQueryStepHandler(msg *db.Message) { + //spew.Dump("SET QUERYSTEP") + qs := msg.Data.(query.SetQueryStep) + result, _ := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId, qs.Bitmap.Filter) + + result_message := db.Message{Data: query.SetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) ClearQueryStepHandler(msg *db.Message) { + //spew.Dump("SET QUERYSTEP") + qs := msg.Data.(query.ClearQueryStep) + result, _ := self.Index.ClearBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) + + result_message := db.Message{Data: query.ClearQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) RangeQueryStepHandler(msg *db.Message) { + qs := msg.Data.(query.RangeQueryStep) + //spew.Dump("RANDE QUERYSTEP") + + bh, err := self.Index.Range(qs.Location.FragmentId, qs.Bitmap.Id, qs.Start, qs.End) + if err != nil { + spew.Dump(err) + } + + var result interface{} + if qs.LocIsDest() { + result = bh + } else { + bm, err := self.Index.GetBytes(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + result = bm + } + result_message := db.Message{Data: query.RangeQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + +func (self *Executor) StashQueryStepHandler(msg *db.Message) { + log.Trace("StashQueryStepHandler") + qs := msg.Data.(query.StashQueryStep) + + part := make(chan interface{}) + num_parts := len(qs.Inputs) + + for _, input := range qs.Inputs { + go func(id *util.GUID, part chan interface{}) { + value, _ := self.Hold.Get(id, util.TimeOut) + part <- value + }(input, part) + } + //just collect all the handles and return them + result := query.NewStash() //query.Stash{make([]query.CacheItem, 0), false} + for i := 0; i < num_parts; i++ { + value := <-part + + switch val := value.(type) { + case index.BitmapHandle: + log.Info("STASH ADDING HANDLE", val) + //not sure what to do here.... + //result.Handles = append(result.Handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + item := query.CacheItem{qs.Location.FragmentId, bh} + result.Stash = append(result.Stash, item) + case query.Stash: + result.Stash = append(result.Stash, val.Stash...) + default: + log.Warn("UNEXCPECTED MESSAGE", value) + } + } + result_message := db.Message{Data: query.StashQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) + } func (self *Executor) RunQueryTest(database_name string, pql string) string { @@ -72,7 +524,7 @@ func (self *Executor) RunQueryTest(database_name string, pql string) string { func (self *Executor) runQuery(database *db.Database, qry *query.Query) error { log.Trace("Executor.runQuery", database, qry) - process, err := self.service.GetProcess() + process, err := self.ProcessMap.GetProcess(&self.ID) if err != nil { return err } @@ -84,9 +536,9 @@ func (self *Executor) runQuery(database *db.Database, qry *query.Query) error { if err != nil { switch obj := err.(type) { case *query.FragmentNotFound: - self.service.TopologyMapper.MakeFragments(obj.Db, obj.Slice) + self.TopologyMapper.MakeFragments(obj.Db, obj.Slice) } - self.service.Hold.Set(qry.Id, err, 30) + self.Hold.Set(qry.Id, err, 30) return err } // loop over the query steps and send to Transport @@ -97,7 +549,7 @@ func (self *Executor) runQuery(database *db.Database, qry *query.Query) error { case query.PortableQueryStep: loc := step.GetLocation() if loc != nil { - self.service.Transport.Send(msg, loc.ProcessId) + self.Transport.Send(msg, loc.ProcessId) } else { log.Warn("Problem with querystep(nil location)", spew.Sdump(step)) } @@ -108,7 +560,7 @@ func (self *Executor) runQuery(database *db.Database, qry *query.Query) error { func (self *Executor) RunPQL(database_name string, pql string) (interface{}, error) { log.Trace("Executor.RunPQL", database_name, pql) - database := self.service.Cluster.GetOrCreateDatabase(database_name) + database := self.Cluster.GetOrCreateDatabase(database_name) // see if the outer query function is a custom query reserved_functions := stringSlice{"get", "set", "clear", "union", "intersect", "difference", "count", "top-n", "mask", "range", "stash", "recall"} @@ -127,15 +579,14 @@ func (self *Executor) RunPQL(database_name string, pql string) (interface{}, err go self.runQuery(database, qry) var final interface{} - final, err = self.service.Hold.Get(qry.Id, 10) + final, err = self.Hold.Get(qry.Id, 10) if err != nil { return nil, err } return final, nil } else { //want to refactor this down to just RunPlugin(tokens) - plugins_dir := config.GetString("plugins") - plugins_file := plugins_dir + "/" + outer_token + ".js" + plugins_file := self.PluginsPath + "/" + outer_token + ".js" filter, filters := query.TokensToFilterStrings(tokens) query_list := GetPlugin(plugins_file, filter, filters).(query.PqlList) @@ -162,7 +613,7 @@ func (self *Executor) RunPQL(database_name string, pql string) (interface{}, err label string err error }) { - final, err := self.service.Hold.Get(q.Id, 10) + final, err := self.Hold.Get(q.Id, 10) result <- struct { final interface{} label string @@ -184,11 +635,121 @@ func (self *Executor) RunPQL(database_name string, pql string) (interface{}, err } +func init() { + gob.Register(TopNPackage{}) + gob.Register(TopFill{}) +} + +type TopNPackage struct { + ProcessId util.GUID + FragmentId util.SUUID + Pairs []index.Pair + HBitmap index.BitmapHandle +} + +type TopFill struct { + Args []index.FillArgs + ReturnProcessId util.GUID + QueryId util.GUID + DestProcessId util.GUID +} + +type Task struct { + processid util.GUID + f map[util.SUUID]index.FillArgs + hold_id util.GUID +} + +func newtask(p util.GUID) *Task { + result := new(Task) + result.processid = p + result.f = make(map[util.SUUID]index.FillArgs) + result.hold_id = util.RandomUUID() + return result +} + +func (t *Task) Add(frag util.SUUID, bitmap_id uint64, handle index.BitmapHandle) { + fa, ok := t.f[frag] + if !ok { + fa = index.FillArgs{frag, handle, make([]uint64, 0, 0)} + } + fa.Bitmaps = append(fa.Bitmaps, bitmap_id) + t.f[frag] = fa +} + +func BuildTask(merge_map map[uint64]uint64, + slice_map map[uint64]map[util.SUUID]struct{}, + total_fragments map[util.SUUID]struct { + process util.GUID + handle index.BitmapHandle + }) map[util.GUID]*Task { + + tasks := make(map[util.GUID]*Task) + for bitmap_id, _ := range merge_map { //for all brands + //for fragment_id, reported_fragments := range slice_map[bitmap_id] { //find missing fragments + reporting_fragments := slice_map[bitmap_id] + //id slice ==> SUUID,BitmapHandle + for _, p := range missing(reporting_fragments, total_fragments) { + task, ok := tasks[p.process] + if !ok { + task = newtask(p.process) + tasks[p.process] = task + } + task.Add(p.fragment, bitmap_id, p.handle) + + } + //} + } + + return tasks +} + +type hole struct { + process util.GUID + handle index.BitmapHandle + fragment util.SUUID +} + +func missing(fids map[util.SUUID]struct{}, all map[util.SUUID]struct { + process util.GUID + handle index.BitmapHandle +}) []hole { + results := make([]hole, 0, 0) + + for k, v := range all { + _, ok := fids[k] + if !ok { + results = append(results, hole{v.process, v.handle, k}) + } + } + return results +} + +func (self *TopFill) GetId() *util.GUID { + return &self.QueryId +} +func (self *TopFill) GetLocation() *db.Location { + return &db.Location{&self.DestProcessId, 0} //this message is a broadcast to many fragments so i'm choosing fragmentzero +} + func (self *Executor) Run() { log.Warn("Executor Run...") } -func NewExecutor(service *core.Service) *Executor { - log.Trace("NewExector") - return &Executor{service, make(chan *db.Message)} +type stringSlice []string + +func (slice stringSlice) pos(value string) int { + for p, v := range slice { + if v == value { + return p + } + } + return -1 +} + +func min(a, b int) int { + if a < b { + return a + } + return b } diff --git a/index/bitmap_test.go b/index/bitmap_test.go index e242ad2e0..e46b7a597 100644 --- a/index/bitmap_test.go +++ b/index/bitmap_test.go @@ -4,7 +4,7 @@ import ( "testing" ) -func TestBitmaps(t *testing.T) { +func TestRBBitmap_SetBit(t *testing.T) { bm := CreateRBBitmap() SetBit(bm, 0) ClearBit(bm, 0) @@ -12,135 +12,3 @@ func TestBitmaps(t *testing.T) { t.Error("Should be 0") } } - -/* -func TestBitmaps(t *testing.T) { - Convey("function BitCount should equal method bm.Count()", t, func() { - bm := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(bm, i) - } - bc1 := BitCount(bm) - bc2 := bm.Count() - So(bc1, ShouldEqual, bc2) - So(bc1, ShouldEqual, 4096) - }) - Convey("function Difference 1 and not 0 => true ", t, func() { - bm1 := CreateRBBitmap() - bm2 := CreateRBBitmap() - SetBit(bm1, 1) - //SetBit(bm2,2) - all := Difference(bm1, bm2) - res := BitCount(all) - - So(1, ShouldEqual, res) - }) - - Convey("function Difference 1 and not 0 => true ", t, func() { - bm1 := CreateRBBitmap() - bm2 := CreateRBBitmap() - SetBit(bm1, 1) - SetBit(bm1, 2) - SetBit(bm1, 3) - SetBit(bm1, 4) - SetBit(bm2, 3) - //SetBit(bm2,2) - all := Difference(bm1, bm2) - res := BitCount(all) - - So(3, ShouldEqual, res) - }) - - Convey("UNION even + odd equal 4096 ", t, func() { - even := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i += 2 { - SetBit(even, i) - } - - odd := CreateRBBitmap() - for i := uint64(1); i < uint64(4096); i += 2 { - SetBit(odd, i) - } - all := Union(even, odd) - total_bits := BitCount(all) - - So(total_bits, ShouldEqual, 4096) - }) - - Convey("Intersection even - odd equal 0 ", t, func() { - even := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i += 2 { - SetBit(even, i) - } - - odd := CreateRBBitmap() - for i := uint64(1); i < uint64(4096); i += 2 { - SetBit(odd, i) - } - all := Intersection(even, odd) - total_bits := BitCount(all) - - So(total_bits, ShouldEqual, 0) - - }) - - Convey("Bitcount< 1s ", t, func() { - all := CreateRBBitmap() - for i := uint64(0); i < uint64(65536); i++ { - SetBit(all, i) - } - start := time.Now() - BitCount(all) - So(start, ShouldHappenWithin, time.Duration(1)*time.Millisecond, time.Now()) - }) - - Convey("Compressed ", t, func() { - all := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(all, i) - } - cs := all.ToCompressString() - fmt.Println(cs) - bm := CreateRBBitmap() - bm.FromCompressString(cs) - So(BitCount(all), ShouldEqual, BitCount(bm)) - }) - - Convey("AndCount ", t, func() { - a := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(a, i) - } - b := CreateRBBitmap() - for i := uint64(0); i < uint64(8192); i++ { - SetBit(b, i) - } - c1 := IntersectionCount(a, b) - c := Intersection(a, b) - So(c1, ShouldEqual, BitCount(c)) - }) - -} -func benchmark(b *testing.B, size int, fill int) { - x := make(map[uint64]IBitmap) - for i := uint64(0); i < uint64(size); i++ { - x[i] = CreateRBBitmap() - } - for i := 0; i < b.N; i++ { - bid := rand.Int() % size - SetBit(x[uint64(bid)], uint64(i%fill)) - } -} -func BenchmarkSetBitL2(b *testing.B) { - benchmark(b, 50000, 1024*64) -} - -/* -func BenchmarkSetBit(b *testing.B) { - // run the Fib function b.N times - a := CreateRBBitmap() - for n := 0; n < b.N; n++ { - SetBit(a, uint64(n)) - } -} -*/ diff --git a/index/brand.go b/index/brand.go index 4a42ea926..7bc6651e4 100644 --- a/index/brand.go +++ b/index/brand.go @@ -10,10 +10,11 @@ import ( log "github.com/cihub/seelog" _ "github.com/go-sql-driver/mysql" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/util" ) +var FragmentBase string + var globalLock *sync.Mutex func init() { @@ -346,7 +347,7 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair { return packagePairs(results[:end]) } func (self *Brand) getFileName() string { - base := config.GetString("fragment_base") + base := FragmentBase if base == "" { base = "." } diff --git a/index/brand_test.go b/index/brand_test.go index aaf3ecf80..59aaddaa7 100644 --- a/index/brand_test.go +++ b/index/brand_test.go @@ -25,117 +25,7 @@ func init() { } } -/* -func TestBitmaps(t *testing.T) { - Convey("function BitCount should equal method bm.Count()", t, func() { - bm := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(bm, i) - } - bc1 := BitCount(bm) - bc2 := bm.Count() - So(bc1, ShouldEqual, bc2) - So(bc1, ShouldEqual, 4096) - }) - Convey("function Difference 1 and not 0 => true ", t, func() { - bm1 := CreateRBBitmap() - bm2 := CreateRBBitmap() - SetBit(bm1, 1) - //SetBit(bm2,2) - all := Difference(bm1, bm2) - res := BitCount(all) - - So(1, ShouldEqual, res) - }) - - Convey("function Difference 1 and not 0 => true ", t, func() { - bm1 := CreateRBBitmap() - bm2 := CreateRBBitmap() - SetBit(bm1, 1) - SetBit(bm1, 2) - SetBit(bm1, 3) - SetBit(bm1, 4) - SetBit(bm2, 3) - //SetBit(bm2,2) - all := Difference(bm1, bm2) - res := BitCount(all) - - So(3, ShouldEqual, res) - }) - - Convey("UNION even + odd equal 4096 ", t, func() { - even := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i += 2 { - SetBit(even, i) - } - - odd := CreateRBBitmap() - for i := uint64(1); i < uint64(4096); i += 2 { - SetBit(odd, i) - } - all := Union(even, odd) - total_bits := BitCount(all) - - So(total_bits, ShouldEqual, 4096) - }) - - Convey("Intersection even - odd equal 0 ", t, func() { - even := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i += 2 { - SetBit(even, i) - } - - odd := CreateRBBitmap() - for i := uint64(1); i < uint64(4096); i += 2 { - SetBit(odd, i) - } - all := Intersection(even, odd) - total_bits := BitCount(all) - - So(total_bits, ShouldEqual, 0) - - }) - - Convey("Bitcount< 1s ", t, func() { - all := CreateRBBitmap() - for i := uint64(0); i < uint64(65536); i++ { - SetBit(all, i) - } - start := time.Now() - BitCount(all) - So(start, ShouldHappenWithin, time.Duration(1)*time.Millisecond, time.Now()) - }) - - Convey("Compressed ", t, func() { - all := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(all, i) - } - cs := all.ToCompressString() - fmt.Println(cs) - bm := CreateRBBitmap() - bm.FromCompressString(cs) - So(BitCount(all), ShouldEqual, BitCount(bm)) - }) - - Convey("AndCount ", t, func() { - a := CreateRBBitmap() - for i := uint64(0); i < uint64(4096); i++ { - SetBit(a, i) - } - b := CreateRBBitmap() - for i := uint64(0); i < uint64(8192); i++ { - SetBit(b, i) - } - c1 := IntersectionCount(a, b) - c := Intersection(a, b) - So(c1, ShouldEqual, BitCount(c)) - }) - -} -*/ -func benchmark_(b *testing.B, size int, fill int, brand *Brand) { - +func benchmarkBrand(b *testing.B, size int, fill int, brand *Brand) { println(b.N) for i := 0; i < b.N; i++ { bid := rand.Int() % size @@ -143,19 +33,6 @@ func benchmark_(b *testing.B, size int, fill int, brand *Brand) { brand.SetBit(uint64(bid), profile, 1) } } -func BenchmarkBrandMemSetBitL2(b *testing.B) { - benchmark_(b, size, 1024*64, membrand) -} -func BenchmarkBrandCasSetBitL2(b *testing.B) { - benchmark_(b, size, 1024*64, cassbrand) -} -/* -func BenchmarkSetBit(b *testing.B) { - // run the Fib function b.N times - a := CreateRBBitmap() - for n := 0; n < b.N; n++ { - SetBit(a, uint64(n)) - } -} -*/ +func BenchmarkBrandMemSetBitL2(b *testing.B) { benchmarkBrand(b, size, 1024*64, membrand) } +func BenchmarkBrandCasSetBitL2(b *testing.B) { benchmarkBrand(b, size, 1024*64, cassbrand) } diff --git a/index/fragment_container.go b/index/fragment_container.go index f757f7830..082bb92cf 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -13,10 +13,16 @@ import ( log "github.com/cihub/seelog" _ "github.com/go-sql-driver/mysql" "github.com/golang/groupcache/lru" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/util" ) +// DefaultBackend is the default data storage layer. +const DefaultBackend = "cassandra" + +var Backend = DefaultBackend + +var LevelDBPath string + type FragmentContainer struct { fragments map[util.SUUID]*Fragment mutex *sync.Mutex @@ -364,14 +370,11 @@ type Fragment struct { } func getStorage(db string, slice int, frame string, fid util.SUUID) Storage { - storage_method := config.GetString("storage_backend") - - switch storage_method { + switch Backend { default: return NewMemoryStorage() case "leveldb": - base_path := config.GetString("level_db_path") - full_dir := fmt.Sprintf("%s/%s/%d/%s/%s", base_path, db, slice, frame, util.SUUID_to_Hex(fid)) + full_dir := fmt.Sprintf("%s/%s/%d/%s/%s", LevelDBPath, db, slice, frame, util.SUUID_to_Hex(fid)) return NewLevelDBStorage(full_dir) case "cassandra": return NewCassStorage() diff --git a/index/fragment_container_test.go b/index/fragment_container_test.go index 082ac7f9d..f22dbaee4 100644 --- a/index/fragment_container_test.go +++ b/index/fragment_container_test.go @@ -1,14 +1,11 @@ package index +/* import ( "testing" - - . "github.com/smartystreets/goconvey/convey" - "github.com/umbel/pilosa/util" ) func TestFragment(t *testing.T) { - general := util.Hex_to_SUUID("1") brand := util.Hex_to_SUUID("2") dummy := NewFragmentContainer() @@ -100,37 +97,36 @@ func TestFragment(t *testing.T) { So(1, ShouldEqual, 1) }) - /* - Convey("Brand SetBit Big", t, func() { - bi1 := uint64(1231) - bi2 := uint64(1232) - bi3 := uint64(1233) - bi4 := uint64(1234) - for x := uint64(0); x < 60000; x++ { - if x < 100 { - dummy.SetBit(brand, bi1, x) - dummy.SetBit(brand, bi4, x) - } - if x < 500 { - dummy.SetBit(brand, bi2, x) - } - if x%3 == 0 && x < 1000 { - dummy.SetBit(brand, bi3, x) - } - if x > 700 && x < 1000 { - dummy.SetBit(brand, bi4, x) - } - if x > 1000 { - dummy.SetBit(brand, x, x) - } - } - bh1, _ := dummy.Get(brand, bi1) - // dummy.Rank() - log.Println(dummy.TopN(brand, bh1, 4)) - log.Println(dummy.Stats(brand)) - So(1, ShouldEqual, 1) - }) - */ + // Convey("Brand SetBit Big", t, func() { + // bi1 := uint64(1231) + // bi2 := uint64(1232) + // bi3 := uint64(1233) + // bi4 := uint64(1234) + // for x := uint64(0); x < 60000; x++ { + // if x < 100 { + // dummy.SetBit(brand, bi1, x) + // dummy.SetBit(brand, bi4, x) + // } + // if x < 500 { + // dummy.SetBit(brand, bi2, x) + // } + // if x%3 == 0 && x < 1000 { + // dummy.SetBit(brand, bi3, x) + // } + // if x > 700 && x < 1000 { + // dummy.SetBit(brand, bi4, x) + // } + // if x > 1000 { + // dummy.SetBit(brand, x, x) + // } + // } + // bh1, _ := dummy.Get(brand, bi1) + // // dummy.Rank() + // log.Println(dummy.TopN(brand, bh1, 4)) + // log.Println(dummy.Stats(brand)) + // So(1, ShouldEqual, 1) + // }) + Convey("Brand TopN", t, func() { dummy.SetBit(brand, uint64(1), 1, 2) dummy.SetBit(brand, uint64(1), 2, 2) @@ -159,3 +155,4 @@ func TestFragment(t *testing.T) { }) } +*/ diff --git a/index/general.go b/index/general.go index 82cf0ca29..e9aba8f25 100644 --- a/index/general.go +++ b/index/general.go @@ -6,7 +6,6 @@ import ( log "github.com/cihub/seelog" "github.com/golang/groupcache/lru" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/util" ) @@ -82,7 +81,7 @@ func (self *General) Stats() interface{} { } func (self *General) getFileName() string { - base := config.GetString("fragment_base") + base := FragmentBase if base == "" { base = "." } diff --git a/index/storage_cass.go b/index/storage_cass.go index b1de73677..d2170bec1 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -7,10 +7,23 @@ import ( log "github.com/cihub/seelog" "github.com/gocql/gocql" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/util" ) +// DefaultStorageHosts are the hosts that stores the data. +var DefaultStorageHosts = [...]string{"localhost"} + +// DefaultStorageKeyspace is the default keyspace used for storage. +const DefaultStorageKeyspace = "pilosa" + +const DefaultCassandraTimeWindow = 5 * time.Second +const DefaultCassandraMaxSizeBatch = 15 + +var StorageHosts = DefaultStorageHosts[:] +var StorageKeyspace = DefaultStorageKeyspace +var CassandraTimeWindow = DefaultCassandraTimeWindow +var CassandraMaxSizeBatch = DefaultCassandraMaxSizeBatch + type CassandraStorage struct { db *gocql.Session batch *gocql.Batch @@ -27,8 +40,8 @@ var session *gocql.Session func SetupCassandra() { var err error - hosts := config.GetStringArrayDefault("cassandra_hosts", []string{"localhost"}) - keyspace := config.GetStringDefault("cassandra_keyspace", "pilosa") + hosts := StorageHosts + keyspace := StorageKeyspace cluster = gocql.NewCluster(hosts...) cluster.Keyspace = keyspace cluster.Consistency = gocql.One @@ -67,8 +80,8 @@ func NewCassStorage() Storage { obj.batch = nil obj.batch_time = time.Now() obj.batch_counter = 0 - obj.cass_time_window_secs = float64(config.GetIntDefault("cassandra_time_window_secs", 5)) - obj.cass_flush_size = config.GetIntDefault("cassandra_max_size_batch", 15) + obj.cass_time_window_secs = float64(CassandraTimeWindow.Seconds()) + obj.cass_flush_size = CassandraMaxSizeBatch return obj } diff --git a/index/storage_test.go b/index/storage_test.go index b099d3c83..617ff10f3 100644 --- a/index/storage_test.go +++ b/index/storage_test.go @@ -1,5 +1,6 @@ package index +/* import ( "fmt" "net" @@ -18,20 +19,20 @@ func TestStorage(t *testing.T) { slice := 0 filter := 10 bitmap_id := uint64(999999) - /* Convey("KV ", t, func() { - storage, _ := NewKVStorage("/tmp/", 0, db) - bm := storage.Fetch(bitmap_id, db, slice) - SetBit(bm, 0) - SetBit(bm, 1) - SetBit(bm, 2) - storage.Store(int64(bitmap_id), db, frame, slice, filter, bm.(*Bitmap)) - bm2, _ := storage.Fetch(bitmap_id, db, slice) - So(BitCount(bm), ShouldEqual, BitCount(bm2)) - So(BitCount(bm), ShouldEqual, bm.Count()) - So(BitCount(bm), ShouldEqual, 3) - }) - */ + // Convey("KV ", t, func() { + // storage, _ := NewKVStorage("/tmp/", 0, db) + // bm := storage.Fetch(bitmap_id, db, slice) + // SetBit(bm, 0) + // SetBit(bm, 1) + // SetBit(bm, 2) + // storage.Store(int64(bitmap_id), db, frame, slice, filter, bm.(*Bitmap)) + // bm2, _ := storage.Fetch(bitmap_id, db, slice) + // So(BitCount(bm), ShouldEqual, BitCount(bm2)) + // So(BitCount(bm), ShouldEqual, bm.Count()) + // So(BitCount(bm), ShouldEqual, 3) + // }) + c, err := net.DialTimeout("tcp", "127.0.0.1:9042", 100*time.Millisecond) if err != nil { fmt.Println("NO cassandra. Skipping test.") @@ -57,27 +58,25 @@ func TestStorage(t *testing.T) { }) } - /* - Convey("leveldb", t, func() { - storage := NewLevelDBStorage("./basic/one") - fmt.Println("FETCH") - bm, _ := storage.Fetch(bitmap_id, db, frame, slice) - //spew.Dump(bm) - SetBit(bm, 0) - SetBit(bm, 1) - SetBit(bm, 2) - fmt.Println("STORE") - storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap)) - //storage.FlushBatch() - fmt.Println("FETCH") - bm2, _ := storage.Fetch(bitmap_id, db, frame, slice) - So(BitCount(bm), ShouldEqual, BitCount(bm2)) - So(BitCount(bm), ShouldEqual, bm.Count()) - So(BitCount(bm), ShouldEqual, 3) - storage.Close() - - }) - */ + // Convey("leveldb", t, func() { + // storage := NewLevelDBStorage("./basic/one") + // fmt.Println("FETCH") + // bm, _ := storage.Fetch(bitmap_id, db, frame, slice) + // //spew.Dump(bm) + // SetBit(bm, 0) + // SetBit(bm, 1) + // SetBit(bm, 2) + // fmt.Println("STORE") + // storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap)) + // //storage.FlushBatch() + // fmt.Println("FETCH") + // bm2, _ := storage.Fetch(bitmap_id, db, frame, slice) + // So(BitCount(bm), ShouldEqual, BitCount(bm2)) + // So(BitCount(bm), ShouldEqual, bm.Count()) + // So(BitCount(bm), ShouldEqual, 3) + // storage.Close() + // }) } +*/ diff --git a/index/timeframe_test.go b/index/timeframe_test.go index 74297d8cb..546f48b9d 100644 --- a/index/timeframe_test.go +++ b/index/timeframe_test.go @@ -1,5 +1,6 @@ package index +/* import ( "fmt" "log" @@ -129,3 +130,4 @@ func TestTimeFrame(t *testing.T) { }) } +*/ diff --git a/transport/http.go b/transport/http.go index dc62e3ed4..eacbe1a74 100644 --- a/transport/http.go +++ b/transport/http.go @@ -6,6 +6,8 @@ import ( "github.com/umbel/pilosa/db" ) +const DefaultHTTPPort = 15001 + type HttpTransport struct { port int outbox chan *db.Message diff --git a/transport/tcp.go b/transport/tcp.go index 248579a76..31415b2b6 100644 --- a/transport/tcp.go +++ b/transport/tcp.go @@ -9,34 +9,35 @@ import ( notify "github.com/bitly/go-notify" log "github.com/cihub/seelog" - "github.com/umbel/pilosa/config" "github.com/umbel/pilosa/core" "github.com/umbel/pilosa/db" - . "github.com/umbel/pilosa/util" + "github.com/umbel/pilosa/util" ) +const DefaultTCPPort = 12001 + type connection struct { transport *TcpTransport inbox chan *db.Message outbox chan *db.Message conn *net.Conn - process *GUID + process *util.GUID } type newconnection struct { - id *GUID + id *util.GUID connection *connection } func init() { - gob.Register(GUID{}) + gob.Register(util.GUID{}) } func (self *connection) manage() { BeginManageConnection: for { if self.conn == nil { - process, err := self.transport.service.ProcessMap.GetProcess(self.process) + process, err := self.transport.ProcessMap.GetProcess(self.process) if err != nil { log.Warn("transport/tcp: error getting process, retrying in 2 seconds... ", self.process, err) time.Sleep(2 * time.Second) @@ -51,7 +52,7 @@ BeginManageConnection: } self.conn = &conn go func() { - self.outbox <- &db.Message{self.transport.service.Id} + self.outbox <- &db.Message{self.transport.ID.String()} }() } encoder := gob.NewEncoder(*self.conn) @@ -78,7 +79,7 @@ BeginManageConnection: return } case message := <-self.inbox: - identifier, ok := message.Data.(GUID) + identifier, ok := message.Data.(util.GUID) if ok { // message is connection registration; bypass inbox and register self.process = &identifier @@ -99,12 +100,26 @@ BeginManageConnection: } type TcpTransport struct { - service *core.Service - port int inbox chan *db.Message outbox chan db.Envelope - connections map[GUID]*connection + connections map[util.GUID]*connection reg chan *newconnection + + ID util.GUID + Port int + ProcessMap *core.ProcessMap +} + +func NewTcpTransport(id util.GUID) *TcpTransport { + return &TcpTransport{ + inbox: make(chan *db.Message, 100), + outbox: make(chan db.Envelope, 100), + connections: make(map[util.GUID]*connection), + reg: make(chan *newconnection), + + ID: id, + Port: DefaultTCPPort, + } } func (self *TcpTransport) Run() { @@ -127,10 +142,10 @@ func (self *TcpTransport) Run() { } func (self *TcpTransport) listen() { - port_string := fmt.Sprintf(":%d", self.port) + port_string := fmt.Sprintf(":%d", self.Port) l, e := net.Listen("tcp", port_string) if e != nil { - log.Critical("Cannot bind to port! ", self.port) + log.Critical("Cannot bind to port! ", self.Port) os.Exit(-1) } for { @@ -153,7 +168,7 @@ func (self *TcpTransport) Close() { log.Warn("Shutting down TCP transport") } -func (self *TcpTransport) Send(message *db.Message, host *GUID) { +func (self *TcpTransport) Send(message *db.Message, host *util.GUID) { log.Trace("TcpTransport.Send", message, host) envelope := db.Envelope{message, host} notify.Post("outbox", &envelope) @@ -170,7 +185,3 @@ func (self *TcpTransport) Receive() *db.Message { func (self *TcpTransport) Push(message *db.Message) { self.inbox <- message } - -func NewTcpTransport(service *core.Service) *TcpTransport { - return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan db.Envelope, 100), make(map[GUID]*connection), make(chan *newconnection)} -} diff --git a/util/id.go b/util/id.go index 109ff6a69..fb3fc1deb 100644 --- a/util/id.go +++ b/util/id.go @@ -63,14 +63,18 @@ func Hex_to_SUUID(str string) SUUID { type GUID [16]byte -func Equal(a, b *GUID) bool { - for i, v := range a { - if v != b[i] { - return false - } +// UnmarshalText parses a text value into a GUID. +// This is used by the TOML parser. +func (id *GUID) UnmarshalText(text []byte) error { + v, err := ParseGUID(string(text)) + if err != nil { + return err } - return true + + *id = v + return nil } + func (self GUID) String() string { var offsets = [...]int{0, 2, 4, 6, 9, 11, 14, 16, 19, 21, 24, 26, 28, 30, 32, 34} const hexString = "0123456789abcdef" @@ -87,12 +91,22 @@ func (self GUID) String() string { } +func Equal(a, b *GUID) bool { + for i, v := range a { + if v != b[i] { + return false + } + } + return true +} + func RandomUUID() GUID { uid, _ := gocql.RandomUUID() var r GUID copy(r[:], uid[:]) return r } + func ParseGUID(input string) (GUID, error) { var u GUID j := 0 diff --git a/util/statd.go b/util/statd.go index bb4bb2481..24951495f 100644 --- a/util/statd.go +++ b/util/statd.go @@ -5,9 +5,14 @@ import ( "github.com/cactus/go-statsd-client/statsd" log "github.com/cihub/seelog" - "github.com/umbel/pilosa/config" ) +// DefaultStatsdHost is the default host to send statsd data to. +const DefaultStatsdHost = "127.0.0.1:8125" + +// StatsdHost is the host to send statsd data to. +var StatsdHost = DefaultStatsdHost + type args struct { stat string delta int64 @@ -20,14 +25,14 @@ var ( end chan bool ) -func SetupUtil() { - setup_storage() +func SetupStatsd() { timer = make(chan args, 32768) count = make(chan string, 32768) end = make(chan bool) - stat_config := config.GetStringDefault("statsd_server", "127.0.0.1:8125") - log.Warn("New Stats", stat_config) - stats, _ := statsd.New(stat_config, "") + + log.Warn("New Stats", StatsdHost) + stats, _ := statsd.New(StatsdHost, "") + go func() { for { select { @@ -50,6 +55,7 @@ func SendTimer(stat string, delta int64) { milli := time.Duration(delta) / time.Millisecond timer <- args{pstat, int64(milli), 1.0} } + func SendInc(stat string) { pstat := "pilosa." + stat count <- pstat diff --git a/util/storage.go b/util/storage.go index 1a766fd86..eace2a47f 100644 --- a/util/storage.go +++ b/util/storage.go @@ -6,17 +6,8 @@ import ( "strings" "github.com/kr/s3/s3util" - "github.com/umbel/pilosa/config" ) -func setup_storage() { - access_key := config.GetString("AWS_ACCESS_KEY_ID") - secret := config.GetString("AWS_SECRET_ACCESS_KEY") - - s3util.DefaultConfig.AccessKey = access_key - s3util.DefaultConfig.SecretKey = secret -} - func Open(s string) (io.ReadCloser, error) { if isURL(s) { return s3util.Open(s, nil)