diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 7393a51fa..648b5cb6d 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -5,26 +5,16 @@ import ( "flag" "fmt" "io" - "io/ioutil" - "log" "math/rand" - "net" - "net/http" - "net/url" "os" "os/signal" "os/user" "path/filepath" - "runtime/pprof" - "strconv" "strings" - "sync" "time" "github.com/BurntSushi/toml" - "github.com/gogo/protobuf/proto" "github.com/umbel/pilosa" - "github.com/umbel/pilosa/internal" ) // Build holds the build information passed in at compile time. @@ -44,9 +34,6 @@ const ( // DefaultHost is the default hostname and port to use. DefaultHost = "localhost:15000" - - // DefaultAntiEntropyInterval is the default interval to run AAE. - DefaultAntiEntropyInterval = 10 * time.Minute ) func main() { @@ -81,27 +68,11 @@ func main() { // Main represents the main program execution. type Main struct { - index *pilosa.Index - ln net.Listener - ticker *time.Ticker - pollingSecs int - - // Close management. - wg sync.WaitGroup - closing chan struct{} - - // Path to the configuration file. - ConfigPath string + Server *pilosa.Server // Configuration options. - Config *Config - - // Cluster configuration shared by components - Host string - Cluster *pilosa.Cluster - - // Profiling paths - CPUProfile string + ConfigPath string + Config *Config // Standard input/output Stdin io.Reader @@ -112,23 +83,15 @@ type Main struct { // NewMain returns a new instance of Main. func NewMain() *Main { return &Main{ - closing: make(chan struct{}), - + Server: pilosa.NewServer(), Config: NewConfig(), + Stdin: os.Stdin, Stdout: os.Stdout, Stderr: os.Stderr, } } -// Addr returns the address of the listener. -func (m *Main) Addr() net.Addr { - if m.ln == nil { - return nil - } - return m.ln.Addr() -} - // Run executes the main program execution. func (m *Main) Run(args ...string) error { // Notify user of config file. @@ -136,193 +99,30 @@ func (m *Main) Run(args ...string) error { fmt.Fprintf(m.Stdout, "Using config: %s\n", m.ConfigPath) } - // Require a port in the hostname. - host, port, err := net.SplitHostPort(m.Config.Host) - if err != nil { - return err - } else if port == "" { - return errors.New("port must be specified in config host") - } + // Setup logging output. + m.Server.LogOutput = m.Stderr - // Set up profiling. - if m.CPUProfile != "" { - f, err := os.Create(m.CPUProfile) - if err != nil { - return err - } - - pprof.StartCPUProfile(f) - defer pprof.StopCPUProfile() - } - - // Open HTTP listener to determine port (if specified as :0). - ln, err := net.Listen("tcp", ":"+port) - if err != nil { - return err - } - m.ln = ln - - // Determine hostname based on listening port. - m.Host = net.JoinHostPort(host, strconv.Itoa(m.ln.Addr().(*net.TCPAddr).Port)) - - // Build cluster from config file. Create local host if none are specified. - m.Cluster = m.Config.PilosaCluster() - if len(m.Cluster.Nodes) == 0 { - m.Cluster.Nodes = []*pilosa.Node{{ - Host: m.Host, - }} - } - - // Create index to store fragments. + // Configure index. fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir) - m.index = pilosa.NewIndex(m.Config.DataDir) - if err := m.index.Open(); err != nil { + m.Server.Index.Path = m.Config.DataDir + + // Build cluster from config file. + m.Server.Host = m.Config.Host + m.Server.Cluster = m.Config.PilosaCluster() + + // Initialize server. + if err := m.Server.Open(); err != nil { return err } - // Create executor for executing queries. - e := pilosa.NewExecutor(m.index) - e.Host = m.Host - e.Cluster = m.Cluster - - // Initialize HTTP handler. - h := pilosa.NewHandler() - h.Index = m.index - h.Host = m.Host - h.Cluster = m.Cluster - h.Executor = e - h.LogOutput = m.Stderr - - // Serve HTTP. - go func() { http.Serve(ln, h) }() - - // Start anti-entropy background worker. - m.wg.Add(1) - go func() { defer m.wg.Done(); m.monitorAntiEntropy() }() - - // Sync up max slice if more than one node - if len(m.Cluster.Nodes) > 1 { - m.ticker = time.NewTicker(time.Second * time.Duration(m.pollingSecs)) - go func() { - for range m.ticker.C { - oldmax := m.index.SliceN() - newmax := oldmax - for _, node := range m.Cluster.Nodes { - if m.Host != node.Host { - newslice, _ := checkMaxSlice(node.Host) - if newslice > newmax { - newmax = newslice - } - } - } - if newmax > oldmax { - m.index.SetMax(newmax) - } - } - }() - } - - fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Host) + fmt.Fprintf(m.Stderr, "Listening as http://%s\n", m.Server.Host) return nil } -func (m *Main) monitorAntiEntropy() { - ticker := time.NewTicker(time.Duration(m.Config.AntiEntropy.Interval)) - defer ticker.Stop() - - m.logger().Printf("index sync monitor initializing") - - for { - // Wait for tick or a close. - select { - case <-m.closing: - return - case <-ticker.C: - } - - m.logger().Printf("index sync beginning") - - // Initialize syncer with local index and remote client. - var syncer pilosa.IndexSyncer - syncer.Index = m.index - syncer.Host = m.Host - syncer.Cluster = m.Cluster - - // Sync indexes. - if err := syncer.SyncIndex(); err != nil { - m.logger().Printf("index sync error: err=%s", err) - continue - } - - // Record successful sync in log. - m.logger().Printf("index sync complete") - } -} - -func checkMaxSlice(hostport string) (uint64, error) { - // Create HTTP request. - req, err := http.NewRequest("GET", (&url.URL{ - Scheme: "http", - Host: hostport, - Path: "/slices/max", - }).String(), nil) - - if err != nil { - return 0, err - } - - // Require protobuf encoding. - req.Header.Set("Accept", "application/x-protobuf") - req.Header.Set("Content-Type", "application/x-protobuf") - - // Send request to remote node. - resp, err := http.DefaultClient.Do(req) - if err != nil { - return 0, err - } - defer resp.Body.Close() - - // Read response into buffer. - body, err := ioutil.ReadAll(resp.Body) - if err != nil { - return 0, err - } - - // Check status code. - if resp.StatusCode != http.StatusOK { - return 0, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) - } - - // Decode response object. - pb := internal.SliceMaxResponse{} - - if err = proto.Unmarshal(body, &pb); err != nil { - return 0, err - } - - return *pb.SliceMax, nil - -} - -// Close shuts down the process. +// Close shuts down the server. func (m *Main) Close() error { - // Notify goroutines to stop. - close(m.closing) - m.wg.Wait() - - if m.ticker != nil { - m.ticker.Stop() - } - if m.ln != nil { - m.ln.Close() - } - - if m.index != nil { - m.index.Close() - } - - return nil + return m.Server.Close() } // ParseFlags parses command line flags from args. @@ -330,8 +130,6 @@ func (m *Main) ParseFlags(args []string) error { fs := flag.NewFlagSet("pilosa", flag.ContinueOnError) fs.SetOutput(m.Stderr) fs.StringVar(&m.ConfigPath, "config", "", "config path") - fs.StringVar(&m.CPUProfile, "cpuprofile", "", "write cpu profile to file") - fs.IntVar(&m.pollingSecs, "pollingSecs", 60, "number of seconds to poll the cluster for maxslice") if err := fs.Parse(args); err != nil { return err } @@ -363,16 +161,15 @@ func (m *Main) ParseFlags(args []string) error { return nil } -func (m *Main) logger() *log.Logger { return log.New(m.Stderr, "", log.LstdFlags) } - // Config represents the configuration for the command. type Config struct { DataDir string `toml:"data-dir"` Host string `toml:"host"` Cluster struct { - ReplicaN int `toml:"replicas"` - Nodes []*ConfigNode `toml:"node"` + ReplicaN int `toml:"replicas"` + Nodes []*ConfigNode `toml:"node"` + PollingInterval Duration `toml:"polling-interval"` } `toml:"cluster"` Plugins struct { @@ -394,7 +191,8 @@ func NewConfig() *Config { Host: DefaultHost, } c.Cluster.ReplicaN = pilosa.DefaultReplicaN - c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) + c.Cluster.PollingInterval = Duration(pilosa.DefaultPollingInterval) + c.AntiEntropy.Interval = Duration(pilosa.DefaultAntiEntropyInterval) return c } diff --git a/cmd/pilosa/main_test.go b/cmd/pilosa/main_test.go index 5ce82a1ef..9bbca9d4f 100644 --- a/cmd/pilosa/main_test.go +++ b/cmd/pilosa/main_test.go @@ -183,11 +183,11 @@ func TestMain_FrameRestore(t *testing.T) { defer m1.Close() // Update cluster config. - m0.Cluster.Nodes = []*pilosa.Node{ - {Host: m0.Host}, - {Host: m1.Host}, + m0.Server.Cluster.Nodes = []*pilosa.Node{ + {Host: m0.Server.Host}, + {Host: m1.Server.Host}, } - m1.Cluster.Nodes = m0.Cluster.Nodes + m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes // Write data on first cluster. if _, err := m0.Query("db=d", ` @@ -214,10 +214,10 @@ func TestMain_FrameRestore(t *testing.T) { defer m2.Close() // Import from first cluster. - client, err := pilosa.NewClient(m2.Host) + client, err := pilosa.NewClient(m2.Server.Host) if err != nil { t.Fatal(err) - } else if err := client.RestoreFrame(m0.Host, "d", "f"); err != nil { + } else if err := client.RestoreFrame(m0.Server.Host, "d", "f"); err != nil { t.Fatal(err) } @@ -324,7 +324,7 @@ func (m *Main) Reopen() error { } // URL returns the base URL string for accessing the running program. -func (m *Main) URL() string { return "http://" + m.Addr().String() } +func (m *Main) URL() string { return "http://" + m.Server.Addr().String() } // Query executes a query against the program through the HTTP API. func (m *Main) Query(rawQuery, query string) (string, error) { diff --git a/executor.go b/executor.go index e46ed30a9..9e46258b8 100644 --- a/executor.go +++ b/executor.go @@ -20,7 +20,7 @@ const DefaultFrame = "general" // Executor recursively executes calls in a PQL query across all slices. type Executor struct { - index *Index + Index *Index // Local hostname & cluster configuration. Host string @@ -31,16 +31,12 @@ type Executor struct { } // NewExecutor returns a new instance of Executor. -func NewExecutor(index *Index) *Executor { +func NewExecutor() *Executor { return &Executor{ - index: index, HTTPClient: http.DefaultClient, } } -// Index returns the index that the executor runs against. -func (e *Executor) Index() *Index { return e.index } - // Execute executes a PQL query. func (e *Executor) Execute(db string, q *pql.Query, slices []uint64, opt *ExecOptions) ([]interface{}, error) { // Verify that a database is set. @@ -56,7 +52,7 @@ func (e *Executor) Execute(db string, q *pql.Query, slices []uint64, opt *ExecOp // If slices aren't specified, then include all of them. if len(slices) == 0 { // Round up the number of slices. - sliceN := e.index.SliceN() + sliceN := e.Index.SliceN() sliceN += (sliceN % uint64(len(e.Cluster.Nodes))) + uint64(len(e.Cluster.Nodes)) // Generate a slices of all slices. @@ -129,7 +125,7 @@ func (e *Executor) executeBitmapCall(db string, c pql.BitmapCall, slices []uint6 // Attach bitmap attributes for Bitmap() calls. if c, ok := c.(*pql.Bitmap); ok { - fr := e.Index().Frame(db, c.Frame) + fr := e.Index.Frame(db, c.Frame) if fr != nil { attrs, err := fr.BitmapAttrStore().Attrs(c.ID) if err != nil { @@ -237,7 +233,7 @@ func (e *Executor) executeTopNSlice(db string, c *pql.TopN, slice uint64) ([]Pai frame = DefaultFrame } - f := e.Index().Fragment(db, frame, slice) + f := e.Index.Fragment(db, frame, slice) if f == nil { return nil, nil } @@ -276,7 +272,7 @@ func (e *Executor) executeBitmapSlice(db string, c *pql.Bitmap, slice uint64) (* frame = DefaultFrame } - f := e.Index().Fragment(db, frame, slice) + f := e.Index.Fragment(db, frame, slice) if f == nil { return NewBitmap(), nil } @@ -309,7 +305,7 @@ func (e *Executor) executeRangeSlice(db string, c *pql.Range, slice uint64) (*Bi frame = DefaultFrame } - f := e.Index().Fragment(db, frame, slice) + f := e.Index.Fragment(db, frame, slice) if f == nil { return NewBitmap(), nil } @@ -364,7 +360,7 @@ func (e *Executor) executeCount(db string, c *pql.Count, slices []uint64, opt *E // executeProfile executes a Profile() call. // This call only executes locally since the profile attibutes are stored locally. func (e *Executor) executeProfile(db string, c *pql.Profile, opt *ExecOptions) (*Profile, error) { - panic("FIXME: impl: e.Index().ProfileAttr(c.ID)") + panic("FIXME: impl: e.Index.ProfileAttr(c.ID)") } // executeClearBit executes a ClearBit() call. @@ -374,7 +370,7 @@ func (e *Executor) executeClearBit(db string, c *pql.ClearBit, opt *ExecOptions) for _, node := range e.Cluster.SliceNodes(slice) { // Update locally if host matches. if node.Host == e.Host { - f, err := e.Index().CreateFragmentIfNotExists(db, c.Frame, slice) + f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice) if err != nil { return false, fmt.Errorf("fragment: %s", err) } @@ -406,7 +402,7 @@ func (e *Executor) executeSetBit(db string, c *pql.SetBit, opt *ExecOptions) (bo for _, node := range e.Cluster.SliceNodes(slice) { // Update locally if host matches. if node.Host == e.Host { - f, err := e.Index().CreateFragmentIfNotExists(db, c.Frame, slice) + f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice) if err != nil { return false, fmt.Errorf("fragment: %s", err) } @@ -438,7 +434,7 @@ func (e *Executor) executeSetBit(db string, c *pql.SetBit, opt *ExecOptions) (bo // executeSetBitmapAttrs executes a SetBitmapAttrs() call. func (e *Executor) executeSetBitmapAttrs(db string, c *pql.SetBitmapAttrs) error { // Retrieve frame. - frame, err := e.Index().CreateFrameIfNotExists(db, c.Frame) + frame, err := e.Index.CreateFrameIfNotExists(db, c.Frame) if err != nil { return err } @@ -456,7 +452,7 @@ func (e *Executor) executeSetBitmapAttrs(db string, c *pql.SetBitmapAttrs) error // executeSetProfileAttrs executes a SetProfileAttrs() call. func (e *Executor) executeSetProfileAttrs(db string, c *pql.SetProfileAttrs) error { // Retrieve database. - d, err := e.Index().CreateDBIfNotExists(db) + d, err := e.Index.CreateDBIfNotExists(db) if err != nil { return err } @@ -490,6 +486,7 @@ func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64, op } // Create HTTP request. + println("dbg.host?", node.Host) req, err := http.NewRequest("POST", (&url.URL{ Scheme: "http", Host: node.Host, diff --git a/executor_test.go b/executor_test.go index 88c4dca33..b7af82276 100644 --- a/executor_test.go +++ b/executor_test.go @@ -460,7 +460,8 @@ type Executor struct { // NewExecutor returns a new instance of Executor. // The executor always matches the hostname of the first cluster node. func NewExecutor(index *pilosa.Index, cluster *pilosa.Cluster) *Executor { - e := &Executor{Executor: pilosa.NewExecutor(index)} + e := &Executor{Executor: pilosa.NewExecutor()} + e.Index = index e.Cluster = cluster e.Host = cluster.Nodes[0].Host return e diff --git a/index.go b/index.go index c09a47ac3..3cd890968 100644 --- a/index.go +++ b/index.go @@ -12,17 +12,18 @@ import ( // Index represents a container for fragments. type Index struct { mu sync.Mutex - path string remoteMax uint64 // Databases by name. dbs map[string]*DB + + // Data directory path. + Path string } // NewIndex returns a new instance of Index. -func NewIndex(path string) *Index { +func NewIndex() *Index { return &Index{ - path: path, dbs: make(map[string]*DB), remoteMax: 0, } @@ -30,12 +31,12 @@ func NewIndex(path string) *Index { // Open initializes the root data directory for the index. func (i *Index) Open() error { - if err := os.MkdirAll(i.path, 0777); err != nil { + if err := os.MkdirAll(i.Path, 0777); err != nil { return err } // Open path to read all database directories. - f, err := os.Open(i.path) + f, err := os.Open(i.Path) if err != nil { return err } @@ -68,9 +69,6 @@ func (i *Index) Close() error { return nil } -// Path returns the path the index was initialized with. -func (i *Index) Path() string { return i.path } - // SliceN returns the highest slice across all frames. func (i *Index) SliceN() uint64 { i.mu.Lock() @@ -101,7 +99,7 @@ func (i *Index) Schema() []*DBInfo { } // DBPath returns the path where a given database is stored. -func (i *Index) DBPath(name string) string { return filepath.Join(i.path, name) } +func (i *Index) DBPath(name string) string { return filepath.Join(i.Path, name) } // DB returns the database by name. func (i *Index) DB(name string) *DB { diff --git a/index_test.go b/index_test.go index a19253592..1c5cdc208 100644 --- a/index_test.go +++ b/index_test.go @@ -25,7 +25,8 @@ func TestIndexSyncer_SyncIndex(t *testing.T) { defer s.Close() s.Handler.Index = idx1.Index s.Handler.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) { - e := pilosa.NewExecutor(idx1.Index) + e := pilosa.NewExecutor() + e.Index = idx1.Index e.Host = cluster.Nodes[1].Host e.Cluster = cluster return e.Execute(db, query, slices, opt) @@ -125,7 +126,10 @@ func NewIndex() *Index { if err != nil { panic(err) } - return &Index{Index: pilosa.NewIndex(path)} + + i := &Index{Index: pilosa.NewIndex()} + i.Path = path + return i } // MustOpenIndex creates and opens an index at a temporary path. Panic on error. @@ -139,7 +143,7 @@ func MustOpenIndex() *Index { // Close closes the index and removes all underlying data. func (i *Index) Close() error { - defer os.RemoveAll(i.Path()) + defer os.RemoveAll(i.Path) return i.Index.Close() } diff --git a/server.go b/server.go new file mode 100644 index 000000000..92047d35c --- /dev/null +++ b/server.go @@ -0,0 +1,259 @@ +package pilosa + +import ( + "errors" + "fmt" + "io" + "io/ioutil" + "log" + "net" + "net/http" + "net/url" + "os" + "strconv" + "sync" + "time" + + "github.com/gogo/protobuf/proto" + "github.com/umbel/pilosa/internal" +) + +// Default server settings. +const ( + DefaultAntiEntropyInterval = 10 * time.Minute + DefaultPollingInterval = 60 * time.Second +) + +// Server represents an index wrapped by a running HTTP server. +type Server struct { + ln net.Listener + + // Close management. + wg sync.WaitGroup + closing chan struct{} + + // Data storage and HTTP interface. + Index *Index + Handler *Handler + + // Cluster configuration. + // Host is replaced with actual host after opening if port is ":0". + Host string + Cluster *Cluster + + // Background monitoring intervals. + AntiEntropyInterval time.Duration + PollingInterval time.Duration + + LogOutput io.Writer +} + +// NewServer returns a new instance of Server. +func NewServer() *Server { + s := &Server{ + closing: make(chan struct{}), + + Index: NewIndex(), + Handler: NewHandler(), + + AntiEntropyInterval: DefaultAntiEntropyInterval, + PollingInterval: DefaultPollingInterval, + + LogOutput: os.Stderr, + } + + s.Handler.Index = s.Index + + return s +} + +// Open opens and initializes the server. +func (s *Server) Open() error { + // Require a port in the hostname. + host, port, err := net.SplitHostPort(s.Host) + if err != nil { + return err + } else if port == "" { + return errors.New("port must be specified in config host") + } + + // Open HTTP listener to determine port (if specified as :0). + ln, err := net.Listen("tcp", ":"+port) + if err != nil { + return err + } + s.ln = ln + + // Determine hostname based on listening port. + s.Host = net.JoinHostPort(host, strconv.Itoa(s.ln.Addr().(*net.TCPAddr).Port)) + + // Create local node if no cluster is specified. + if len(s.Cluster.Nodes) == 0 { + s.Cluster.Nodes = []*Node{{Host: s.Host}} + } + + // Open index. + if err := s.Index.Open(); err != nil { + return err + } + + // Create executor for executing queries. + e := NewExecutor() + e.Index = s.Index + e.Host = s.Host + e.Cluster = s.Cluster + + // Initialize HTTP handler. + s.Handler.Host = s.Host + s.Handler.Cluster = s.Cluster + s.Handler.Executor = e + s.Handler.LogOutput = s.LogOutput + + // Serve HTTP. + go func() { http.Serve(ln, s.Handler) }() + + // Start background monitoring. + s.wg.Add(2) + go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() + go func() { defer s.wg.Done(); s.monitorMaxSlice() }() + + return nil +} + +// Close closes the server and waits for it to shutdown. +func (s *Server) Close() error { + // Notify goroutines to stop. + close(s.closing) + s.wg.Wait() + + if s.ln != nil { + s.ln.Close() + } + if s.Index != nil { + s.Index.Close() + } + + return nil +} + +// Addr returns the address of the listener. +func (s *Server) Addr() net.Addr { + if s.ln == nil { + return nil + } + return s.ln.Addr() +} + +func (s *Server) logger() *log.Logger { return log.New(s.LogOutput, "", log.LstdFlags) } + +func (s *Server) monitorAntiEntropy() { + ticker := time.NewTicker(time.Duration(s.AntiEntropyInterval)) + defer ticker.Stop() + + s.logger().Printf("index sync monitor initializing") + + for { + // Wait for tick or a close. + select { + case <-s.closing: + return + case <-ticker.C: + } + + s.logger().Printf("index sync beginning") + + // Initialize syncer with local index and remote client. + var syncer IndexSyncer + syncer.Index = s.Index + syncer.Host = s.Host + syncer.Cluster = s.Cluster + + // Sync indexes. + if err := syncer.SyncIndex(); err != nil { + s.logger().Printf("index sync error: err=%s", err) + continue + } + + // Record successful sync in log. + s.logger().Printf("index sync complete") + } +} + +// monitorMaxSlice periodically pulls the highest slice from each node in the cluster. +func (s *Server) monitorMaxSlice() { + // Ignore if only one node in the cluster. + if len(s.Cluster.Nodes) <= 1 { + return + } + + ticker := time.NewTicker(time.Second * time.Duration(s.PollingInterval)) + defer ticker.Stop() + + for { + select { + case <-s.closing: + return + case <-ticker.C: + } + + oldmax := s.Index.SliceN() + newmax := oldmax + for _, node := range s.Cluster.Nodes { + if s.Host != node.Host { + newslice, _ := checkMaxSlice(node.Host) + if newslice > newmax { + newmax = newslice + } + } + } + + if newmax > oldmax { + s.Index.SetMax(newmax) + } + } +} + +func checkMaxSlice(hostport string) (uint64, error) { + // Create HTTP request. + req, err := http.NewRequest("GET", (&url.URL{ + Scheme: "http", + Host: hostport, + Path: "/slices/max", + }).String(), nil) + + if err != nil { + return 0, err + } + + // Require protobuf encoding. + req.Header.Set("Accept", "application/x-protobuf") + req.Header.Set("Content-Type", "application/x-protobuf") + + // Send request to remote node. + resp, err := http.DefaultClient.Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + + // Read response into buffer. + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return 0, err + } + + // Check status code. + if resp.StatusCode != http.StatusOK { + return 0, fmt.Errorf("invalid status: code=%d, err=%s", resp.StatusCode, body) + } + + // Decode response object. + pb := internal.SliceMaxResponse{} + + if err = proto.Unmarshal(body, &pb); err != nil { + return 0, err + } + + return *pb.SliceMax, nil + +}