LogDog: Add Butler stream server package.

client/internal/logdog/butler/streamserver/handshake.go

+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package streamserver
+
+import (
+        "bytes"
+        "encoding/json"
+        "errors"
+        "fmt"
+        "io"
+
+        "github.com/luci/luci-go/client/logdog/butlerlib/streamproto"
+        "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/iotools"
+        "github.com/luci/luci-go/common/logdog/protocol"
+        log "github.com/luci/luci-go/common/logging"
+        "github.com/luci/luci-go/common/proto/google"
+        "github.com/luci/luci-go/common/recordio"
+        "golang.org/x/net/context"
+)
+
+type handshakeProtocol interface {
+        // Handshake executes the protocol handshake, advancing the underlying
+        // stream and returning the derived configuration.
+        //
+        // If the configured stream is invalid, an error will be returned.
+        Handshake(context.Context, io.Reader) (*streamproto.Properties, error)
+}
+
+// handshakeProtocolImpl is a Butler handshake protocol that uses a JSON blob to
+// describe the stream.
+type handshakeProtocolImpl struct {
+        forceVerbose bool // (Testing) force verbose code path.
+}
+
+const (
+        // The maximum size of the header (1MB).
+        maxHeaderSize = 1 * 1024 * 1024
+)
+
+// Reads header/magic bytes from the stream. Returns a protocol instance based
+// on the content of the magic number. If the magic number is invalid, an error
+// will be returned.
+func getProtocolFromMagic(ctx context.Context, r io.Reader) (handshakeProtocol, error) {

[iannucci, 2015/11/12 22:44:24]

why is this whole thing split into two functions? Why not just

  handshake(context.Context, io.Reader) (*streamproto.Properties, error)

?

Is it 'in case you have 2 protocol versions'? If that's the case then I would
strongly advocate just splitting it up when you have a second version.

[dnj, 2015/11/13 20:17:14]

Essentially yes. Okay, SGTM Done.

+        // Read the frame header magic number (version)
+        magic := make([]byte, len(streamproto.ProtocolFrameHeaderMagic))
+        if n, err := io.ReadFull(r, magic); (n != len(magic)) || (err != nil) {
+                return nil, fmt.Errorf("Failed to read frame header magic number: %s", err)
+        }
+
+        // Check the magic number/version
+        if bytes.Equal(magic, streamproto.ProtocolFrameHeaderMagic) {
+                return &handshakeProtocolImpl{}, nil
+        }
+
+        log.Errorf(log.SetField(ctx, "magic", fmt.Sprintf("%#X", magic)),
+                "Unrecognized frame header magic number.")
+        return nil, errors.New("streamserver: Unknown protocol magic in frame header")
+}
+
+func (p *handshakeProtocolImpl) defaultFlags() *streamproto.Flags {
+        return &streamproto.Flags{
+                Type: streamproto.StreamType(protocol.LogStreamDescriptor_TEXT),
+                Tee:  streamproto.TeeNone,
+        }
+}
+
+func (p *handshakeProtocolImpl) Handshake(ctx context.Context, r io.Reader) (*streamproto.Properties, error) {
+        // Load the JSON into our descriptor field.
+        flags, err := p.loadFlags(ctx, r)
+        if err != nil {
+                return nil, err
+        }
+
+        props := flags.Properties()
+        if props.Timestamp == nil {
+                props.Timestamp = google.NewTimestamp(clock.Now(ctx))
+        }
+        if err := props.Validate(); err != nil {
+                return nil, err
+        }
+
+        return props, nil
+}
+
+func (p *handshakeProtocolImpl) loadFlags(ctx context.Context, r io.Reader) (*streamproto.Flags, error) {
+        fr := recordio.NewReader(r, maxHeaderSize)
+
+        // Read the header frame.
+        count, hr, err := fr.ReadFrame()

[iannucci, 2015/11/12 22:44:24]

s/count/frameSize ?

[dnj, 2015/11/13 20:17:14]

Done.

+        if err != nil {
+                log.Errorf(log.SetError(ctx, err), "Failed to read header frame.")
+                return nil, err
+        }
+
+        // When tracing, buffer the JSON data locally so we can emit it via log.
+        headerBuf := bytes.Buffer{}
+        captureHeader := log.IsLogging(ctx, log.Debug) || p.forceVerbose
+        if captureHeader {
+                hr = io.TeeReader(hr, &headerBuf)
+        }
+
+        // When we hand the header reader to the "json" library, we want to count how
+        // many bytes it reads from it. We will assert that it has read the full set
+        // of bytes.
+        chr := &iotools.CountingReader{Reader: hr}
+
+        // Decode into our protocol description structure. Note that extra fields
+        // are ignored (no error) and missing fields retain their zero value.
+        f := p.defaultFlags()
+        err = json.NewDecoder(chr).Decode(f)
+        if captureHeader {
+                log.Fields{
+                        "size":       count,
+                        "decodeSize": headerBuf.Len(),
+                }.Debugf(ctx, "Read JSON header:\n%s", headerBuf.String())
+        }
+        if err != nil {
+                log.Errorf(log.SetError(ctx, err),
+                        "Failed to decode stream description data JSON.")
+                return nil, err
+        }
+
+        // Make sure that this consumed the full JSON size that was specified.
+        //
+        // We use a countReader because the 'json' library doesn't give us a way to
+        // know how many bytes it consumed when it decoded.
+        if chr.Count() != count {
+                return nil, fmt.Errorf("Stream description block was not fully consumed (%d != %d)",
+                        chr.Count(), count)
+        }
+
+        return f, nil
+}