From 71dd6e23f482a268d618dadba5a8cbea4f5f6ac8 Mon Sep 17 00:00:00 2001 From: James DeFelice Date: Sun, 16 Aug 2015 23:41:23 -0400 Subject: [PATCH 1/4] initial revision of recordio decoding --- lib/records/doc.go | 9 ++++++ lib/records/records.go | 56 +++++++++++++++++++++++++++++++++ lib/records/records_test.go | 63 +++++++++++++++++++++++++++++++++++++ 3 files changed, 128 insertions(+) create mode 100644 lib/records/doc.go create mode 100644 lib/records/records.go create mode 100644 lib/records/records_test.go diff --git a/lib/records/doc.go b/lib/records/doc.go new file mode 100644 index 0000000..53049cb --- /dev/null +++ b/lib/records/doc.go @@ -0,0 +1,9 @@ +package records + +// Package records implements the Mesos variant of RecordIO decoding, whereby +// each record is prefixed by a line that indicates the length (decimal, printed +// in ASCII) of the record. The octets of the record immediately follow the +// length-line. Zero-length records are allowed. +// +// This package does not enforce any particular record format: that choice is +// left up to the caller in the form of the Unmarshaler. diff --git a/lib/records/records.go b/lib/records/records.go new file mode 100644 index 0000000..b65239a --- /dev/null +++ b/lib/records/records.go @@ -0,0 +1,56 @@ +package records + +import ( + "bufio" + "io" + "net/textproto" + "strconv" +) + +type Decoder interface { + Decode(v interface{}) error +} + +type Unmarshaler func([]byte, interface{}) error + +type decoder struct { + in *textproto.Reader + un Unmarshaler +} + +var empty = []byte{} + +func NewDecoder(in io.Reader, un Unmarshaler) Decoder { + bio := bufio.NewReader(in) + return &decoder{ + in: textproto.NewReader(bio), + un: un, + } +} + +func (d *decoder) Decode(v interface{}) error { + ll, err := d.in.ReadLine() + if err != nil { + return err + } + + nbytes, err := strconv.ParseInt(ll, 10, 64) + if err != nil { + return err + } + + // zero-length messages are allowed, should we return a secondary param + // to indicate this condition? + if nbytes == 0 { + return d.un(empty, v) + } + + // TODO(jdef) enforce max message size here + + buf := make([]byte, int(nbytes)) + _, err = io.ReadFull(d.in.R, buf) + if err != nil { + return err + } + return d.un(buf, v) +} diff --git a/lib/records/records_test.go b/lib/records/records_test.go new file mode 100644 index 0000000..77df4a9 --- /dev/null +++ b/lib/records/records_test.go @@ -0,0 +1,63 @@ +package records + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" +) + +func ExampleDecoder() { + un := Unmarshaler(func(b []byte, v interface{}) error { + s, ok := v.([]string) + if !ok { + return errors.New("unexpected object store type") + } + if len(s) == 0 { + return errors.New("not enough space in object store") + } + s[0] = string(b) + return nil + }) + d := NewDecoder(bytes.NewBufferString("5\nhello0\n6\nworld!"), un) + s := []string{""} + for { + err := d.Decode(s) + if err != nil { + if err != io.EOF { + fmt.Println(err) + } + break + } else if s[0] == "" { + fmt.Println("--empty--") + } else { + fmt.Println(s[0]) + } + } + // Output: + // hello + // --empty-- + // world! +} + +func ExampleDecoder_json() { + type Demo struct { + Hello string + } + + s := &Demo{} + un := Unmarshaler(json.Unmarshal) + d := NewDecoder(bytes.NewBufferString("18\n{\"hello\": \"world\"}"), un) + + err := d.Decode(s) + if err != nil { + if err != io.EOF { + fmt.Println(err) + } + } else { + fmt.Printf("%+v", *s) + } + // Output: + // {Hello:world} +} From 548f575423225f0edbf7e271cea4cf0f6c4388b8 Mon Sep 17 00:00:00 2001 From: James DeFelice Date: Mon, 17 Aug 2015 02:43:19 -0400 Subject: [PATCH 2/4] refactor lib to use transport pkg --- lib/lib.go | 62 ++---------- lib/records/records.go | 6 ++ lib/transport/transport.go | 201 +++++++++++++++++++++++++++++++++++++ 3 files changed, 217 insertions(+), 52 deletions(-) create mode 100644 lib/transport/transport.go diff --git a/lib/lib.go b/lib/lib.go index 3997d00..9acc9bb 100644 --- a/lib/lib.go +++ b/lib/lib.go @@ -1,16 +1,11 @@ package lib import ( - "bytes" - "encoding/json" - "fmt" - "io" - "io/ioutil" "log" - "net/http" "github.com/gogo/protobuf/proto" "github.com/jimenez/mesoscon-demo/lib/mesosproto" + "github.com/jimenez/mesoscon-demo/lib/transport" ) const ENDPOINT = "/master/api/v1/scheduler" @@ -32,23 +27,13 @@ func New(master, name string) *DemoLib { } } -func (lib *DemoLib) handleEvents(body io.Reader) { - dec := json.NewDecoder(body) - for { - var event mesosproto.Event - - if err := dec.Decode(&event); err != nil { - if err == io.EOF { - break - } - if event.GetType() == mesosproto.Event_UPDATE { - taskStatus := event.GetUpdate().GetStatus() - log.Println("Status for", taskStatus.GetTaskId().GetValue(), "is", taskStatus.GetState().String()) - } - continue - } - +func (lib *DemoLib) handleEvents(s transport.Subscription) { + ech := s.Events() + for event := range ech { switch event.GetType() { + case mesosproto.Event_UPDATE: + taskStatus := event.GetUpdate().GetStatus() + log.Println("Status for", taskStatus.GetTaskId().GetValue(), "is", taskStatus.GetState().String()) case mesosproto.Event_SUBSCRIBED: lib.frameworkID = event.GetSubscribed().GetFrameworkId() log.Println("framework", lib.name, "subscribed succesfully (", lib.frameworkID.String(), ")") @@ -59,42 +44,15 @@ func (lib *DemoLib) handleEvents(body io.Reader) { log.Println("framework", lib.name, "received", len(event.GetOffers().GetOffers()), "offer(s)") } } + log.Println("subscription terminated:", s.Err()) } func (lib *DemoLib) Subscribe() error { - call := mesosproto.Call{ - Type: mesosproto.Call_SUBSCRIBE.Enum(), - Subscribe: &mesosproto.Call_Subscribe{ - FrameworkInfo: lib.frameworkInfo, - }, - } - - body, err := proto.Marshal(&call) + s, err := transport.Subscribe(lib.master, lib.frameworkInfo) if err != nil { return err } - - req, err := http.NewRequest("POST", "http://"+lib.master+ENDPOINT, bytes.NewReader(body)) - if err != nil { - return err - } - req.Header.Set("Content-Type", "application/x-protobuf") - req.Header.Set("Accept", "application/json") - - resp, err := http.DefaultClient.Do(req) - if err != nil { - return err - } - - if resp.StatusCode != 200 { - body, err := ioutil.ReadAll(resp.Body) - if err != nil { - return err - } - return fmt.Errorf("%s", body) - } - - go lib.handleEvents(resp.Body) + go lib.handleEvents(s) return nil } diff --git a/lib/records/records.go b/lib/records/records.go index b65239a..48f3a0c 100644 --- a/lib/records/records.go +++ b/lib/records/records.go @@ -11,6 +11,12 @@ type Decoder interface { Decode(v interface{}) error } +type DecoderFunc func(v interface{}) error + +func (df DecoderFunc) Decode(v interface{}) error { + return df(v) +} + type Unmarshaler func([]byte, interface{}) error type decoder struct { diff --git a/lib/transport/transport.go b/lib/transport/transport.go new file mode 100644 index 0000000..7a118f2 --- /dev/null +++ b/lib/transport/transport.go @@ -0,0 +1,201 @@ +package transport + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "io/ioutil" + "log" + "net/http" + "sync" + "time" + + "github.com/gogo/protobuf/proto" + "github.com/jimenez/mesoscon-demo/lib/mesosproto" + "github.com/jimenez/mesoscon-demo/lib/records" + "golang.org/x/net/context" +) + +type EventChan <-chan *mesosproto.Event + +type Subscription interface { + Events() EventChan // closes when the subscription has terminated + Close() + Err() error // yields the error that caused the Events() channel to close +} + +type subscription struct { + body io.ReadCloser + events chan *mesosproto.Event + cancel context.CancelFunc + err error + errLock sync.Mutex +} + +func (t *subscription) Events() EventChan { + return EventChan(t.events) +} + +func (t *subscription) Close() { + t.cancel() +} + +func (t *subscription) Err() error { + t.errLock.Lock() + defer t.errLock.Unlock() + return t.err +} + +func Subscribe(masterURI string, fi *mesosproto.FrameworkInfo) (Subscription, error) { + call := mesosproto.Call{ + Type: mesosproto.Call_SUBSCRIBE.Enum(), + Subscribe: &mesosproto.Call_Subscribe{ + FrameworkInfo: fi, + }, + } + + body, err := proto.Marshal(&call) + if err != nil { + return nil, err + } + + const EP_SCHEDULER = "/master/api/v1/scheduler" + req, err := http.NewRequest("POST", "http://"+masterURI+EP_SCHEDULER, bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Accept", "application/json") + + var t *subscription + bg := context.Background() + ctx, _ := context.WithTimeout(bg, 75*time.Second) + err = httpDo(ctx, req, func(resp *http.Response, err error) error { + if err != nil { + return err + } + if resp.StatusCode != 200 { + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + return err + } + return fmt.Errorf("%s", body) + } + c2, cancel := context.WithCancel(bg) + t = &subscription{ + body: resp.Body, + events: make(chan *mesosproto.Event, 100), + cancel: cancel, + } + go t.handleEvents(c2) + return nil + }) + return t, err +} + +func httpDo(ctx context.Context, req *http.Request, f func(*http.Response, error) error) error { + // Run the HTTP request in a goroutine and pass the response to f. + tr := &http.Transport{} + client := &http.Client{Transport: tr} + c := make(chan error, 1) + go func() { c <- f(client.Do(req)) }() + select { + case <-ctx.Done(): + tr.CancelRequest(req) + <-c // Wait for f to return. + return ctx.Err() + case err := <-c: + return err + } +} + +func (t *subscription) setError(err error) { + t.errLock.Lock() + defer t.errLock.Unlock() + t.err = err +} + +func (t *subscription) handleEvents(ctx context.Context) { + defer t.cancel() + defer t.body.Close() + defer close(t.events) + + var empty bool + un := records.Unmarshaler(func(b []byte, v interface{}) error { + if b == nil || len(b) == 0 { + empty = true + return nil + } + return json.Unmarshal(b, v) + }) + errNoPulse := errors.New("failed to receive a pulse") + dec := records.NewDecoder(t.body, un) + timed := timedDecoder(dec, 15*time.Second, 5, errNoPulse) + for { + empty = false + event := &mesosproto.Event{} + if err := timed.Decode(event); err != nil { + if err == io.EOF || err == errNoPulse { + t.setError(err) + return + } + // protocol error? log and attempt to continue + // TODO(jdef) not all protocol errors are recoverable + // TODO(jdef) log all of these + log.Println("ERROR:", err) + continue + } + if empty { + // TODO(jdef) lame heartbeat? + continue + } + select { + case t.events <- event: + case <-ctx.Done(): + // no async op to cancel, just abort + t.setError(ctx.Err()) + return + } + } +} + +// timedDecoder returns a Decorated decoder that generates the given error if no events +// are decoded for some number of sequential timeouts. The returned Decoder is not safe +// to share across goroutines. +func timedDecoder(dec records.Decoder, dur time.Duration, timeouts int, err error) records.Decoder { + var t *time.Timer + return records.DecoderFunc(func(v interface{}) error { + if t == nil { + t = time.NewTimer(dur) + } else { + t.Reset(dur) + } + defer t.Stop() + + errCh := make(chan error, 1) + go func() { + // there's no way to abort this so someone else will have + // to make sure that it dies (and it should if the response + // body is closed) + errCh <- dec.Decode(v) + }() + for x := 0; x < timeouts; x++ { + select { + case <-t.C: + // check for a tie + select { + case e := <-errCh: + return e + default: + // noop, continue + } + case e := <-errCh: + return e + } + } + return err + }) +} From 70d0813b96c5da1934bbd2e5635be254c63c26db Mon Sep 17 00:00:00 2001 From: James DeFelice Date: Mon, 17 Aug 2015 10:10:36 -0400 Subject: [PATCH 3/4] refactor decoder into a state machine, gain fidelity w/ respect to underlying stream errors --- lib/records/records.go | 99 +++++++++++++++++++++++++++++++++--------- 1 file changed, 78 insertions(+), 21 deletions(-) diff --git a/lib/records/records.go b/lib/records/records.go index 48f3a0c..70d5fb3 100644 --- a/lib/records/records.go +++ b/lib/records/records.go @@ -2,15 +2,30 @@ package records import ( "bufio" + "fmt" "io" "net/textproto" "strconv" ) +const DefaultMaxEventLength = uint64(1024 * 1024) + +var ( + empty = []byte{} + MaxEventLength = DefaultMaxEventLength // should not be larger than what int will hold +) + type Decoder interface { Decode(v interface{}) error } +// A ProtocolError describes a protocol violation such as an invalid `event-length`. +type ProtocolError string + +func (p ProtocolError) Error() string { + return string(p) +} + type DecoderFunc func(v interface{}) error func (df DecoderFunc) Decode(v interface{}) error { @@ -20,43 +35,85 @@ func (df DecoderFunc) Decode(v interface{}) error { type Unmarshaler func([]byte, interface{}) error type decoder struct { - in *textproto.Reader - un Unmarshaler + in *textproto.Reader + un Unmarshaler + state stateFn + remaining int + records chan []byte + buf []byte } -var empty = []byte{} +type stateFn func(d *decoder) (stateFn, error) func NewDecoder(in io.Reader, un Unmarshaler) Decoder { bio := bufio.NewReader(in) return &decoder{ - in: textproto.NewReader(bio), - un: un, + in: textproto.NewReader(bio), + un: un, + records: make(chan []byte, 1), + state: headerState, + } +} + +func (d *decoder) Decode(v interface{}) (err error) { +stateLoop: + for err == nil { + if d.state, err = d.state(d); err == nil { + select { + case r := <-d.records: + err = d.un(r, v) + break stateLoop + default: + } + } } + return } -func (d *decoder) Decode(v interface{}) error { +func headerState(d *decoder) (stateFn, error) { ll, err := d.in.ReadLine() if err != nil { - return err + //TODO(jdef) check for EOF? + return headerState, err } - - nbytes, err := strconv.ParseInt(ll, 10, 64) + nbytes, err := strconv.ParseUint(ll, 10, 64) if err != nil { - return err + return failedState, ProtocolError(fmt.Sprintf("protocol violation, failed to parse event-length: %v", err)) } - - // zero-length messages are allowed, should we return a secondary param - // to indicate this condition? - if nbytes == 0 { - return d.un(empty, v) + // enforce max message size here + if nbytes > MaxEventLength { + // TODO(jdef) enter a "skip-bytes" state instead? if so, we may need to indicate + // that the error is temporary + return failedState, ProtocolError(fmt.Sprintf("protocol violation, event-length %d exceeds max allowed (%d)", nbytes, MaxEventLength)) + } + if d.remaining = int(nbytes); d.remaining == 0 { + d.buf = empty + } else { + // TODO(jdef) attempt to reuse existing buffer? + d.buf = make([]byte, d.remaining) } + return eventState, nil +} - // TODO(jdef) enforce max message size here +func failedState(d *decoder) (stateFn, error) { + return failedState, ProtocolError("decoder is in a failed state and will not recover") +} - buf := make([]byte, int(nbytes)) - _, err = io.ReadFull(d.in.R, buf) - if err != nil { - return err +func eventState(d *decoder) (stateFn, error) { + var err error + if d.remaining > 0 { + off := len(d.buf) - d.remaining + n, e := d.in.R.Read(d.buf[off:]) + if n > 0 { + d.remaining -= n + } + err = e + } + if d.remaining == 0 { + d.records <- d.buf + // don't communicate an error in the same step that we generated a record, + // let the next stateFn deal with any stream errors (e.g. EOF) + return headerState, nil } - return d.un(buf, v) + return eventState, err } From 2893bb3b534c05b4a95f42ec99c89e62903aa3d5 Mon Sep 17 00:00:00 2001 From: James DeFelice Date: Mon, 17 Aug 2015 12:22:28 -0400 Subject: [PATCH 4/4] add force parameter to subscribe call --- lib/lib.go | 2 +- lib/transport/transport.go | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/lib.go b/lib/lib.go index 9acc9bb..3a16f8a 100644 --- a/lib/lib.go +++ b/lib/lib.go @@ -48,7 +48,7 @@ func (lib *DemoLib) handleEvents(s transport.Subscription) { } func (lib *DemoLib) Subscribe() error { - s, err := transport.Subscribe(lib.master, lib.frameworkInfo) + s, err := transport.Subscribe(lib.master, lib.frameworkInfo, false) if err != nil { return err } diff --git a/lib/transport/transport.go b/lib/transport/transport.go index 7a118f2..ca5f082 100644 --- a/lib/transport/transport.go +++ b/lib/transport/transport.go @@ -48,11 +48,12 @@ func (t *subscription) Err() error { return t.err } -func Subscribe(masterURI string, fi *mesosproto.FrameworkInfo) (Subscription, error) { +func Subscribe(masterURI string, fi *mesosproto.FrameworkInfo, force bool) (Subscription, error) { call := mesosproto.Call{ Type: mesosproto.Call_SUBSCRIBE.Enum(), Subscribe: &mesosproto.Call_Subscribe{ FrameworkInfo: fi, + Force: &force, }, } @@ -165,6 +166,9 @@ func (t *subscription) handleEvents(ctx context.Context) { // timedDecoder returns a Decorated decoder that generates the given error if no events // are decoded for some number of sequential timeouts. The returned Decoder is not safe // to share across goroutines. +// TODO(jdef) this probably isn't the right place for all of this logic (and it's not +// just monitoring the heartbeat messages, it's counting all of them..). Heartbeat monitoring +// has specific requirements. Get rid of this and implement something better elsewhere. func timedDecoder(dec records.Decoder, dur time.Duration, timeouts int, err error) records.Decoder { var t *time.Timer return records.DecoderFunc(func(v interface{}) error {