scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 16:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor main into pilosa.Server

Browse source 
This commit refactors most of the code in `cmd/pilosa` to
`pilosa.Server` so that it can be reused in long running cluster
testing.
This commit is contained in:
Ben Johnson 2016-05-13 14:39:09 -06:00
parent d76aa17e9c
commit 523cf5bc0e
No known key found for this signature in database
GPG key ID: CBD06EAD6DFD9529
7 changed files with 319 additions and 262 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

250
cmd/pilosa/main.go
View file

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

14
cmd/pilosa/main_test.go
View file

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

29
executor.go
View file

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

3
executor_test.go
View file

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

16
index.go
View file

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

10
index_test.go
View file

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

259
server.go Normal file
View file

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