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

Refactor config & service

Browse source 
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.
This commit is contained in:
Ben Johnson 2015-08-12 14:35:24 -06:00
parent c5afd86ff2
commit 2f1e3b6078
38 changed files with 1584 additions and 1464 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

120
cmd/pilosa/config.go Normal file
View file

@ -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
}

190
cmd/pilosa/config_test.go Normal file
View file

@ -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
}

222
cmd/pilosa/main.go Normal file
View file

@ -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
}

1
cmd/pilosa/main_test.go Normal file
View file

@ -0,0 +1 @@
package main_test

28
cmd/pilosa/seelog.go Normal file
View file

@ -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())
//<seelog minlevel="debug" maxlevel="error">
s := fmt.Sprintf(`<seelog minlevel="%s">
<outputs>
<rollingfile type="size" filename="%s" maxsize="524288000" maxrolls="4" formatid="format1" />
</outputs> `, level, fname)
s += `<formats>
<format id="format1" format="%Date/%Time [%LEV] %Msg%n"/>
</formats>
</seelog>`
return s
}

60
commands/pilosa/main.go
View file

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

165
config/config.go
View file

@ -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
}

67
config/config_test.go
View file

@ -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")
})
}

3
config/test.yaml
View file

@ -1,3 +0,0 @@
port_tcp: 12000
port_http: 15000
temp: /tmp

28
core/batch.go
View file

@ -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
}

14
core/core_test.go
View file

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

102
core/etcd.go
View file

@ -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() {

93
core/http.go
View file

@ -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

10
core/load.go
View file

@ -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
}

22
core/ping.go
View file

@ -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)

353
core/query.go
View file

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

21
core/remotebits.go
View file

@ -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)

85
core/service.go
View file

@ -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)
//<seelog minlevel="debug" maxlevel="error">
log_level := config.GetStringDefault("log_level", "info")
prod_config := fmt.Sprintf(`<seelog minlevel="%s">
<outputs>
<rollingfile type="size" filename="%s" maxsize="524288000" maxrolls="4" formatid="format1" />
</outputs> `, log_level, fname)
prod_config += `<formats>
<format id="format1" format="%Date/%Time [%LEV] %Msg%n"/>
</formats>
</seelog>`
//fmt.Println(prod_config)
return prod_config
}
func (self *Service) getDev() string {
return `<seelog>
<outputs>
<console />
</outputs>
</seelog>
`
}
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()

199
core/topn.go
View file

@ -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
}

8
db/topology.go
View file

@ -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

2
default.yaml
View file

@ -14,4 +14,4 @@ supported_frames:
- d
- p.n
etcd_servers:
- http://127.0.0.1:4001
- http://127.0.0.1:4001

5
deps.json
View file

@ -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",

78
dispatch/dispatch.go
View file

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

625
executor/executor.go
View file

@ -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
}

134
index/bitmap_test.go
View file

@ -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))
}
}
*/

5
index/brand.go
View file

@ -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 = "."
}

129
index/brand_test.go
View file

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

15
index/fragment_container.go
View file

@ -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()

67
index/fragment_container_test.go
View file

@ -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) {
})
}
*/

3
index/general.go
View file

@ -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 = "."
}

23
index/storage_cass.go
View file

@ -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
}

67
index/storage_test.go
View file

@ -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()
// })
}
*/

2
index/timeframe_test.go
View file

@ -1,5 +1,6 @@
package index
/*
import (
"fmt"
"log"
@ -129,3 +130,4 @@ func TestTimeFrame(t *testing.T) {
})
}
*/

2
transport/http.go
View file

@ -6,6 +6,8 @@ import (
"github.com/umbel/pilosa/db"
)
const DefaultHTTPPort = 15001
type HttpTransport struct {
port int
outbox chan *db.Message

47
transport/tcp.go
View file

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

26
util/id.go
View file

@ -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

18
util/statd.go
View file

@ -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

9
util/storage.go
View file

@ -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)