LogDog: Add Butler Output package.

client/internal/logdog/butler/output/null/null.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 null
+
+import (
+        "encoding/hex"
+        "strconv"
+        "sync"
+
+        "github.com/golang/protobuf/proto"
+        "github.com/luci/luci-go/client/internal/logdog/butler/output"
+        "github.com/luci/luci-go/common/logdog/protocol"
+        "github.com/luci/luci-go/common/logdog/types"
+        log "github.com/luci/luci-go/common/logging"
+        "golang.org/x/net/context"
+)
+
+// nullOutput is an Output implementation that logs messages to its contexts'
+// Logger instance.
+type nullOutput struct {
+        ctx context.Context
+
+        // bundleSize is the maximum size of the Butler bundle to use.
+        bundleSize int
+
+        stats   output.StatsBase
+        statsMu sync.Mutex
+}
+
+// nullOutput implements output.Output.
+var _ output.Output = (*nullOutput)(nil)
+
+func (o *nullOutput) String() string {
+        return "Null"
+}
+
+// New instantes a new null output instance.
+func New(ctx context.Context, bundleSize int) output.Output {
+        o := nullOutput{
+                ctx:        ctx,
+                bundleSize: bundleSize,
+        }
+        o.ctx = log.SetFields(o.ctx, log.Fields{
+                "output": &o,
+        })
+        return &o
+}
+
+func (o *nullOutput) SendBundle(bundle *protocol.ButlerLogBundle) error {
+        st := output.StatsBase{}
+        defer o.mergeStats(&st)
+
+        for _, e := range bundle.Entries {
+                path := types.StreamName(e.Desc.Prefix).Join(types.StreamName(e.Desc.Name))
+                ctx := log.SetField(o.ctx, "streamPath", path)
+
+                log.Fields{
+                        "count":      len(e.Logs),
+                        "descriptor": e.Desc.String(),
+                }.Infof(ctx, "Received stream logs.")
+
+                for _, le := range e.Logs {
+                        log.Fields{
+                                "timeOffset":  le.TimeOffset.Duration(),
+                                "prefixIndex": le.PrefixIndex,
+                                "streamIndex": le.StreamIndex,
+                                "sequence":    le.Sequence,
+                        }.Infof(ctx, "Received message.")
+                        if c := le.GetText(); c != nil {
+                                for idx, l := range c.Lines {
+                                        log.Infof(ctx, "Line %d) %s (%s)", idx, l.Value, strconv.Quote(l.Delimiter))
+                                }
+                        }
+                        if c := le.GetBinary(); c != nil {
+                                log.Infof(ctx, "Binary) %s", hex.EncodeToString(c.Data))
+                        }
+                        if c := le.GetDatagram(); c != nil {
+                                log.Infof(ctx, "Datagram %d: %s", c.Size, hex.EncodeToString(c.Data))
+                        }
+
+                        st.F.SentMessages++
+                }
+        }
+        st.F.SentBytes += proto.Size(bundle)
+
+        return nil
+}
+
+func (o *nullOutput) MaxSize() int {
+        return o.bundleSize
+}
+
+func (o *nullOutput) Stats() output.Stats {
+        o.statsMu.Lock()
+        defer o.statsMu.Unlock()
+
+        st := o.stats
+        return &st
+}
+
+func (o *nullOutput) Close() {}
+
+func (o *nullOutput) mergeStats(s output.Stats) {
+        o.statsMu.Lock()
+        defer o.statsMu.Unlock()
+
+        o.stats.Merge(s)
+}