Support reading flows from stdin

Implement an ObserverClient that reads jsonpb flows from an io.Reader
instead of connecting to Hubble, and modify `hubble observer` command
to use it when stdin is a pipe. For example,

    hubble observe -o jsonpb --last 1000 > flows.log
    cat flows.log | hubble observe -n kube-system

A few things to note for observing flows from stdin:

- --follow, --last, and server flags are ignored.
- Flows are processed in the order they are read from stdin without
  sorting them by timestamp.

Fixes #523

Signed-off-by: Michi Mutsuzaki <michi@isovalent.com>

cmd/observe/io_reader_observer.go

@@ -0,0 +1,103 @@
+// Copyright 2021 Authors of Hubble
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package observe
+
+import (
+	"bufio"
+	"context"
+	"io"
+
+	"github.com/cilium/cilium/api/v1/observer"
+	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
+	"github.com/cilium/cilium/pkg/hubble/filters"
+	"github.com/cilium/hubble/pkg/logger"
+	"github.com/golang/protobuf/jsonpb"
+	"google.golang.org/grpc"
+)
+
+// ioReaderObserver implements ObserverClient interface. It reads flows
+// in jsonpb format from an io.Reader.
+type ioReaderObserver struct {
+	reader *bufio.Reader
+}
+
+func newIOReaderObserver(reader io.Reader) *ioReaderObserver {
+	return &ioReaderObserver{
+		reader: bufio.NewReader(reader),
+	}
+}
+
+func (o *ioReaderObserver) GetFlows(_ context.Context, in *observer.GetFlowsRequest, _ ...grpc.CallOption) (observer.Observer_GetFlowsClient, error) {
+	return newIOReaderClient(o.reader, in)
+}
+
+func (o *ioReaderObserver) GetNodes(_ context.Context, _ *observer.GetNodesRequest, _ ...grpc.CallOption) (*observer.GetNodesResponse, error) {
+	panic("GetNodes is not supported")
+}
+
+func (o *ioReaderObserver) ServerStatus(_ context.Context, _ *observer.ServerStatusRequest, _ ...grpc.CallOption) (*observer.ServerStatusResponse, error) {
+	panic("ServerStatus is not supported")
+}
+
+// ioReaderClient implements Observer_GetFlowsClient.
+type ioReaderClient struct {
+	reader  *bufio.Reader
+	request *observer.GetFlowsRequest
+	allow   filters.FilterFuncs
+	deny    filters.FilterFuncs
+	grpc.ClientStream
+}
+
+func newIOReaderClient(reader *bufio.Reader, request *observer.GetFlowsRequest) (*ioReaderClient, error) {
+	allow, err := filters.BuildFilterList(context.Background(), request.Whitelist, filters.DefaultFilters)
+	if err != nil {
+		return nil, err
+	}
+	deny, err := filters.BuildFilterList(context.Background(), request.Blacklist, filters.DefaultFilters)
+	if err != nil {
+		return nil, err
+	}
+	return &ioReaderClient{
+		reader:  reader,
+		request: request,
+		allow:   allow,
+		deny:    deny,
+	}, nil
+}
+
+func (c *ioReaderClient) Recv() (*observer.GetFlowsResponse, error) {
+	for {
+		line, err := c.reader.ReadString('\n')
+		if err != nil {
+			return nil, err
+		}
+		var res observer.GetFlowsResponse
+		err = jsonpb.UnmarshalString(line, &res)
+		if err != nil {
+			logger.Logger.WithError(err).WithField("line", line).Warn("Failed to unmarshal json to flow")
+			continue
+		}
+		if c.request.Since != nil && c.request.Since.AsTime().After(res.Time.AsTime()) {
+			continue
+		}
+		if c.request.Until != nil && c.request.Until.AsTime().Before(res.Time.AsTime()) {
+			continue
+		}
+		if !filters.Apply(c.allow, c.deny, &v1.Event{Timestamp: res.Time, Event: res.GetFlow()}) {
+			continue
+		}
+		return &res, nil
+	}
+}

cmd/observe/io_reader_observer_test.go

@@ -0,0 +1,123 @@
+// Copyright 2021 Authors of Hubble
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package observe
+
+import (
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/cilium/cilium/api/v1/flow"
+	"github.com/cilium/cilium/api/v1/observer"
+	"github.com/golang/protobuf/ptypes/timestamp"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_getFlowsBasic(t *testing.T) {
+	flows := []*observer.GetFlowsResponse{{}, {}, {}}
+	var flowStrings []string
+	for _, f := range flows {
+		b, err := f.MarshalJSON()
+		assert.NoError(t, err)
+		flowStrings = append(flowStrings, string(b))
+	}
+	server := newIOReaderObserver(strings.NewReader(strings.Join(flowStrings, "\n") + "\n"))
+	req := observer.GetFlowsRequest{}
+	client, err := server.GetFlows(context.Background(), &req)
+	assert.NoError(t, err)
+	for i := 0; i < len(flows); i++ {
+		_, err = client.Recv()
+		assert.NoError(t, err)
+	}
+	_, err = client.Recv()
+	assert.Equal(t, io.EOF, err)
+}
+
+func Test_getFlowsTimeRange(t *testing.T) {
+	flows := []*observer.GetFlowsResponse{
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_FORWARDED}},
+			Time:          &timestamp.Timestamp{Seconds: 0},
+		},
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_DROPPED}},
+			Time:          &timestamp.Timestamp{Seconds: 100},
+		},
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_ERROR}},
+			Time:          &timestamp.Timestamp{Seconds: 200},
+		},
+	}
+	var flowStrings []string
+	for _, f := range flows {
+		b, err := f.MarshalJSON()
+		assert.NoError(t, err)
+		flowStrings = append(flowStrings, string(b))
+	}
+	server := newIOReaderObserver(strings.NewReader(strings.Join(flowStrings, "\n") + "\n"))
+	req := observer.GetFlowsRequest{
+		Since: &timestamp.Timestamp{Seconds: 50},
+		Until: &timestamp.Timestamp{Seconds: 150},
+	}
+	client, err := server.GetFlows(context.Background(), &req)
+	assert.NoError(t, err)
+	res, err := client.Recv()
+	assert.NoError(t, err)
+	assert.Equal(t, flows[1], res)
+	_, err = client.Recv()
+	assert.Equal(t, io.EOF, err)
+}
+
+func Test_getFlowsFilter(t *testing.T) {
+	flows := []*observer.GetFlowsResponse{
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_FORWARDED}},
+			Time:          &timestamp.Timestamp{Seconds: 0},
+		},
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_DROPPED}},
+			Time:          &timestamp.Timestamp{Seconds: 100},
+		},
+		{
+			ResponseTypes: &observer.GetFlowsResponse_Flow{Flow: &flow.Flow{Verdict: flow.Verdict_ERROR}},
+			Time:          &timestamp.Timestamp{Seconds: 200},
+		},
+	}
+	var flowStrings []string
+	for _, f := range flows {
+		b, err := f.MarshalJSON()
+		assert.NoError(t, err)
+		flowStrings = append(flowStrings, string(b))
+	}
+	server := newIOReaderObserver(strings.NewReader(strings.Join(flowStrings, "\n") + "\n"))
+	req := observer.GetFlowsRequest{
+		Whitelist: []*flow.FlowFilter{
+			{
+				Verdict: []flow.Verdict{flow.Verdict_FORWARDED, flow.Verdict_ERROR},
+			},
+		},
+	}
+	client, err := server.GetFlows(context.Background(), &req)
+	assert.NoError(t, err)
+	res, err := client.Recv()
+	assert.NoError(t, err)
+	assert.Equal(t, flows[0], res)
+	res, err = client.Recv()
+	assert.NoError(t, err)
+	assert.Equal(t, flows[2], res)
+	_, err = client.Recv()
+	assert.Equal(t, io.EOF, err)
+}

cmd/observe/observe.go

@@ -110,6 +110,20 @@ func New(vp *viper.Viper) *cobra.Command {
 
 func newObserveCmd(vp *viper.Viper, ofilter *observeFilter) *cobra.Command {
 	observeCmd := &cobra.Command{
+		Example: `* Piping flows to hubble observe
+
+  Save output from 'hubble observe -o jsonpb' command to a file, and pipe it to
+  the observe command later for offline processing. For example:
+
+    hubble observe -o jsonpb --last 1000 > flows.json
+
+  Then,
+
+    cat flows.json | hubble observe
+
+  Note that the observe command ignores --follow, --last, and server flags when it
+  reads flows from stdin. The observe command processes and output flows in the same
+  order they are read from stdin without sorting them by timestamp.`,
 		Use:   "observe",
 		Short: "Observe flows of a Hubble server",
 		Long: `Observe provides visibility into flow information on the network and
@@ -121,19 +135,32 @@ more.`,
 			if err := handleArgs(ofilter, debug); err != nil {
 				return err
 			}
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
-			hubbleConn, err := conn.New(ctx, vp.GetString(config.KeyServer), vp.GetDuration(config.KeyTimeout))
+			req, err := getRequest(ofilter)
 			if err != nil {
 				return err
 			}
-			defer hubbleConn.Close()
-			req, err := getRequest(ofilter)
+
+			var client observer.ObserverClient
+			fi, err := os.Stdin.Stat()
 			if err != nil {
 				return err
 			}
+			if fi.Mode()&os.ModeNamedPipe != 0 {
+				// read flows from stdin
+				client = newIOReaderObserver(os.Stdin)
+			} else {
+				// read flows from a hubble server
+				ctx, cancel := context.WithCancel(context.Background())
+				defer cancel()
+				hubbleConn, err := conn.New(ctx, vp.GetString(config.KeyServer), vp.GetDuration(config.KeyTimeout))
+				if err != nil {
+					return err
+				}
+				defer hubbleConn.Close()
+				client = observer.NewObserverClient(hubbleConn)
+			}
+
 			logger.Logger.WithField("request", req).Debug("Sending GetFlows request")
-			client := observer.NewObserverClient(hubbleConn)
 			if err := getFlows(client, req); err != nil {
 				msg := err.Error()
 				// extract custom error message from failed grpc call

go.mod

@@ -4,6 +4,7 @@ go 1.16
 
 require (
 	github.com/cilium/cilium v1.9.0-rc1.0.20210209141502-b944040a9ec8
+	github.com/golang/protobuf v1.5.0
 	github.com/google/go-cmp v0.5.5
 	github.com/gordonklaus/ineffassign v0.0.0-20210209182638-d0e41b2fc8ed
 	github.com/sirupsen/logrus v1.7.0