Merge pull request #1593 from safchain/finally-protobuf

Use protobuf between agent/analyzer as first level protocol

Author: lebauce

Makefile

@@ -29,7 +29,7 @@ endef
 
 define PROTOC_GEN
 $(call VENDOR_RUN,${PROTOC_GEN_GOFAST_GITHUB})
-$(call VENDOR_RUN,${PROTOC_GEN_GO_GITHUB}) protoc -Ivendor -I. --plugin=${BUILD_TOOLS}/protoc-gen-gogofaster --gogofaster_out . $1
+$(call VENDOR_RUN,${PROTOC_GEN_GO_GITHUB}) protoc -Ivendor -I. --plugin=${BUILD_TOOLS}/protoc-gen-gogofaster --gogofaster_out $$GOPATH/src $1
 endef
 
 VERSION?=$(shell $(VERSION_CMD))
@@ -211,8 +211,8 @@ debug.analyzer:
 %.pb.go: %.proto
 	$(call PROTOC_GEN,$<)
 
-flow/flow.pb.go: flow/flow.proto
-	$(call PROTOC_GEN,$<)
+flow/flow.pb.go: flow/flow.proto filters/filters.proto
+	$(call PROTOC_GEN,flow/flow.proto)
 
 	# always export flow.ParentUUID as we need to store this information to know
 	# if it's a Outer or Inner packet.
@@ -232,10 +232,17 @@ flow/flow.pb.go: flow/flow.proto
 flow/layers/generated.proto: flow/layers/layers.go
 	$(call VENDOR_RUN,${PROTEUS_GITHUB}) proteus proto -f $${GOPATH}/src -p github.com/skydive-project/skydive/flow/layers
 	sed -e 's/^package .*;/package layers;/' -i $@
+	sed -e 's/^option go_package = "layers"/option go_package = "github.com\/skydive-project\/skydive\/flow\/layers"/' -i $@
 	sed -e 's/^message Layer/message /' -i $@
 	sed -e 's/option (gogoproto.typedecl) = false;//' -i $@
 	sed 's/\((gogoproto\.customname) = "\([^\"]*\)"\)/\1, (gogoproto.jsontag) = "\2,omitempty"/' -i $@
 
+websocket/structmessage.pb.go: websocket/structmessage.proto
+	$(call PROTOC_GEN,$<)
+
+	sed -e 's/type StructMessage struct {/type StructMessage struct { XXX_state structMessageState `json:"-"`/' -i websocket/structmessage.pb.go
+	gofmt -s -w $@
+
 .proto: govendor flow/layers/generated.pb.go flow/flow.pb.go filters/filters.pb.go websocket/structmessage.pb.go
 
 .PHONY: .proto.clean

agent/agent.go

@@ -45,6 +45,7 @@ import (
 	"github.com/skydive-project/skydive/probe"
 	"github.com/skydive-project/skydive/topology"
 	"github.com/skydive-project/skydive/ui"
+	"github.com/skydive-project/skydive/websocket"
 	ws "github.com/skydive-project/skydive/websocket"
 )
 
@@ -67,7 +68,7 @@ type Agent struct {
 
 // NewAnalyzerStructClientPool creates a new http WebSocket client Pool
 // with authentification
-func NewAnalyzerStructClientPool(authOptions *shttp.AuthenticationOpts) (*ws.StructClientPool, error) {
+func NewAnalyzerStructClientPool(authOpts *shttp.AuthenticationOpts) (*ws.StructClientPool, error) {
 	pool := ws.NewStructClientPool("AnalyzerClientPool")
 
 	addresses, err := config.GetAnalyzerServiceAddresses()
@@ -81,7 +82,8 @@ func NewAnalyzerStructClientPool(authOptions *shttp.AuthenticationOpts) (*ws.Str
 	}
 
 	for _, sa := range addresses {
-		c, err := config.NewWSClient(common.AgentService, config.GetURL("ws", sa.Addr, sa.Port, "/ws/agent/topology"), authOptions, nil)
+		url := config.GetURL("ws", sa.Addr, sa.Port, "/ws/agent/topology")
+		c, err := config.NewWSClient(common.AgentService, url, websocket.ClientOpts{AuthOpts: authOpts, Protocol: websocket.ProtobufProtocol})
 		if err != nil {
 			return nil, err
 		}

analyzer/flow_client.go

@@ -110,7 +110,7 @@ func (c *FlowClientWebSocketConn) Close() error {
 
 // Connect to the WebSocket flow server
 func (c *FlowClientWebSocketConn) Connect() (err error) {
-	if c.wsClient, err = config.NewWSClient(common.AgentService, c.url, c.authOpts, nil); err != nil {
+	if c.wsClient, err = config.NewWSClient(common.AgentService, c.url, ws.ClientOpts{AuthOpts: c.authOpts}); err != nil {
 		return nil
 	}
 
@@ -146,7 +146,7 @@ func (c *FlowClient) close() {
 
 // SendFlow sends a flow to the server
 func (c *FlowClient) SendFlow(f *flow.Flow) error {
-	data, err := f.GetData()
+	data, err := f.Marshal()
 	if err != nil {
 		return err
 	}
@@ -164,10 +164,11 @@ retry:
 }
 
 // SendFlows sends flows to the server
-func (c *FlowClient) SendFlows(flows []*flow.Flow) {
-	for _, flow := range flows {
-		err := c.SendFlow(flow)
-		if err != nil {
+func (c *FlowClient) SendFlows(flowArray *flow.FlowArray) {
+	// TODO(safchain) in case of websocket there is no size limitation we should send
+	// bulk directly
+	for _, flow := range flowArray.Flows {
+		if err := c.SendFlow(flow); err != nil {
 			logging.GetLogger().Errorf("Unable to send flow: %s", err)
 		}
 	}
@@ -240,7 +241,7 @@ func (p *FlowClientPool) OnDisconnected(c ws.Speaker) {
 }
 
 // SendFlows sends flows using a random connection
-func (p *FlowClientPool) SendFlows(flows []*flow.Flow) {
+func (p *FlowClientPool) SendFlows(flowArray *flow.FlowArray) {
 	p.RLock()
 	defer p.RUnlock()
 
@@ -249,7 +250,7 @@ func (p *FlowClientPool) SendFlows(flows []*flow.Flow) {
 	}
 
 	fc := p.flowClients[rand.Intn(len(p.flowClients))]
-	fc.SendFlows(flows)
+	fc.SendFlows(flowArray)
 }
 
 // Close all connections

analyzer/flow_server.go

@@ -101,11 +101,15 @@ type FlowServer struct {
 
 // OnMessage event
 func (c *FlowServerWebSocketConn) OnMessage(client ws.Speaker, m ws.Message) {
-	f, err := flow.FromData(m.Bytes(client.GetClientProtocol()))
-	if err != nil {
+	// rawmessage at this point
+	b, _ := m.Bytes(ws.RawProtocol)
+
+	var f flow.Flow
+	if err := f.Unmarshal(b); err != nil {
 		logging.GetLogger().Errorf("Error while parsing flow: %s", err)
 		return
 	}
+
 	logging.GetLogger().Debugf("New flow from Websocket connection: %+v", f)
 	if len(c.ch) >= c.maxFlowBufferSize {
 		c.numOfLostFlows++
@@ -118,7 +122,7 @@ func (c *FlowServerWebSocketConn) OnMessage(client ws.Speaker, m ws.Message) {
 		return
 	}
 
-	c.ch <- f
+	c.ch <- &f
 }
 
 // Serve starts a WebSocket flow server
@@ -161,8 +165,8 @@ func (c *FlowServerUDPConn) Serve(ch chan *flow.Flow, quit chan struct{}, wg *sy
 					logging.GetLogger().Errorf("Error while reading: %s", err)
 				}
 
-				f, err := flow.FromData(data[0:n])
-				if err != nil {
+				var f flow.Flow
+				if err := f.Unmarshal(data[0:n]); err != nil {
 					logging.GetLogger().Errorf("Error while parsing flow: %s", err)
 					continue
 				}
@@ -178,7 +182,7 @@ func (c *FlowServerUDPConn) Serve(ch chan *flow.Flow, quit chan struct{}, wg *sy
 					}
 					return
 				}
-				ch <- f
+				ch <- &f
 			}
 		}
 	}()
@@ -202,13 +206,13 @@ func NewFlowServerUDPConn(addr string, port int) (*FlowServerUDPConn, error) {
 	return &FlowServerUDPConn{conn: conn, maxFlowBufferSize: flowsMax}, err
 }
 
-func (s *FlowServer) storeFlows(flows []*flow.Flow) {
-	if len(flows) > 0 {
+func (s *FlowServer) storeFlows(flows *flow.FlowArray) {
+	if len(flows.Flows) > 0 {
 		if s.storage != nil {
-			if err := s.storage.StoreFlows(flows); err != nil {
+			if err := s.storage.StoreFlows(flows.Flows); err != nil {
 				logging.GetLogger().Error(err)
 			} else {
-				logging.GetLogger().Debugf("%d flows stored", len(flows))
+				logging.GetLogger().Debugf("%d flows stored", len(flows.Flows))
 			}
 		}
 
@@ -228,21 +232,21 @@ func (s *FlowServer) Start() {
 		dlTimer := time.NewTicker(s.bulkInsertDeadline)
 		defer dlTimer.Stop()
 
-		var flowBuffer []*flow.Flow
-		defer s.storeFlows(flowBuffer)
+		var flowArray flow.FlowArray
+		defer s.storeFlows(&flowArray)
 
 		for {
 			select {
 			case <-s.quit:
 				return
 			case <-dlTimer.C:
-				s.storeFlows(flowBuffer)
-				flowBuffer = flowBuffer[:0]
+				s.storeFlows(&flowArray)
+				flowArray.Flows = flowArray.Flows[:0]
 			case f := <-s.ch:
-				flowBuffer = append(flowBuffer, f)
-				if len(flowBuffer) >= s.bulkInsert {
-					s.storeFlows(flowBuffer)
-					flowBuffer = flowBuffer[:0]
+				flowArray.Flows = append(flowArray.Flows, f)
+				if len(flowArray.Flows) >= s.bulkInsert {
+					s.storeFlows(&flowArray)
+					flowArray.Flows = flowArray.Flows[:0]
 				}
 			}
 		}

analyzer/flow_subscriber_endpoint.go

@@ -33,8 +33,8 @@ type FlowSubscriberEndpoint struct {
 }
 
 // SendFlows sends flow to the subscribers
-func (fs *FlowSubscriberEndpoint) SendFlows(flows []*flow.Flow) {
-	msg := ws.NewStructMessage("flow", "store", flows)
+func (fs *FlowSubscriberEndpoint) SendFlows(flowArray *flow.FlowArray) {
+	msg := ws.NewStructMessage("flow", "store", flowArray)
 	fs.pool.BroadcastMessage(msg)
 }
 

api/server/pcap.go

@@ -26,7 +26,7 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/abbot/go-http-auth"
+	auth "github.com/abbot/go-http-auth"
 	"github.com/skydive-project/skydive/config"
 	"github.com/skydive-project/skydive/flow"
 	"github.com/skydive-project/skydive/flow/storage"
@@ -40,10 +40,10 @@ type PcapAPI struct {
 	Storage storage.Storage
 }
 
-func (p *PcapAPI) flowExpireUpdate(flows []*flow.Flow) {
-	if p.Storage != nil && len(flows) > 0 {
-		p.Storage.StoreFlows(flows)
-		logging.GetLogger().Debugf("%d flows stored", len(flows))
+func (p *PcapAPI) flowExpireUpdate(flowArray *flow.FlowArray) {
+	if p.Storage != nil && len(flowArray.Flows) > 0 {
+		p.Storage.StoreFlows(flowArray.Flows)
+		logging.GetLogger().Debugf("%d flows stored", len(flowArray.Flows))
 	}
 }
 

api/server/topology.go

@@ -30,7 +30,7 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/abbot/go-http-auth"
+	auth "github.com/abbot/go-http-auth"
 	"github.com/skydive-project/skydive/api/types"
 	"github.com/skydive-project/skydive/flow"
 	"github.com/skydive-project/skydive/graffiti/graph"
@@ -193,7 +193,7 @@ func (t *TopologyAPI) topologySearch(w http.ResponseWriter, r *auth.Authenticate
 		w.Header().Set("Content-Type", "application/json; charset=UTF-8")
 		w.WriteHeader(http.StatusOK)
 		if err := json.NewEncoder(w).Encode(res); err != nil {
-			logging.GetLogger().Warningf("Error while writing response: %s", err)
+			logging.GetLogger().Errorf("Error while writing response: %s", err)
 		}
 	}
 }

cmd/client/topology.go

@@ -27,7 +27,6 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
-	"net/http"
 	"os"
 
 	"github.com/skydive-project/skydive/common"
@@ -75,9 +74,9 @@ var TopologyImport = &cobra.Command{
 		}
 
 		url := config.GetURL("ws", sa.Addr, sa.Port, "/ws/publisher")
-		headers := http.Header{}
-		headers.Add("X-Persistence-Policy", string(hub.Persistent))
-		client, err := config.NewWSClient(common.UnknownService, url, &AuthenticationOpts, headers)
+		opts := websocket.ClientOpts{AuthOpts: &AuthenticationOpts}
+		opts.Headers.Add("X-Persistence-Policy", string(hub.Persistent))
+		client, err := config.NewWSClient(common.UnknownService, url, opts)
 		if err != nil {
 			exitOnError(err)
 		}

config/websocket.go

@@ -23,7 +23,6 @@
 package config
 
 import (
-	"net/http"
 	"net/url"
 	"time"
 
@@ -33,16 +32,19 @@ import (
 )
 
 // NewWSClient creates a Client based on the configuration
-func NewWSClient(clientType common.ServiceType, url *url.URL, authOpts *shttp.AuthenticationOpts, headers http.Header) (*websocket.Client, error) {
+func NewWSClient(clientType common.ServiceType, url *url.URL, opts websocket.ClientOpts) (*websocket.Client, error) {
 	host := GetString("host_id")
-	queueSize := GetInt("http.ws.queue_size")
-	writeCompression := GetBool("http.ws.enable_write_compression")
+
+	// override some of the options with config value
+	opts.QueueSize = GetInt("http.ws.queue_size")
+	opts.WriteCompression = GetBool("http.ws.enable_write_compression")
 	tlsConfig, err := GetTLSClientConfig(true)
 	if err != nil {
 		return nil, err
 	}
+	opts.TLSConfig = tlsConfig
 
-	return websocket.NewClient(host, clientType, url, authOpts, headers, queueSize, writeCompression, tlsConfig), nil
+	return websocket.NewClient(host, clientType, url, opts), nil
 }
 
 // NewWSServer creates a Server based on the configuration

contrib/objectstore/main.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"net/http"
 	"net/url"
 	"os"
 	"os/signal"
@@ -12,6 +11,7 @@ import (
 	"github.com/skydive-project/skydive/contrib/objectstore/subscriber"
 	shttp "github.com/skydive-project/skydive/http"
 	"github.com/skydive-project/skydive/logging"
+	"github.com/skydive-project/skydive/websocket"
 )
 
 const defaultConfigurationFile = "/etc/skydive/skydive-objectstore.yml"
@@ -40,7 +40,7 @@ func main() {
 	subscriberPassword := cfg.GetString("subscriber_password")
 	maxSecondsPerStream := cfg.GetInt("max_seconds_per_stream")
 
-	authOptions := &shttp.AuthenticationOpts{
+	authOpts := &shttp.AuthenticationOpts{
 		Username: subscriberUsername,
 		Password: subscriberPassword,
 	}
@@ -51,7 +51,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	wsClient, err := config.NewWSClient(common.AnalyzerService, subscriberURL, authOptions, http.Header{})
+	wsClient, err := config.NewWSClient(common.AnalyzerService, subscriberURL, websocket.ClientOpts{AuthOpts: authOpts})
 	if err != nil {
 		logging.GetLogger().Errorf("Failed to create websocket client: %s", err)
 		os.Exit(1)