Merge pull request #1669 from safchain/debug-weird-node-not-found

Fix graph replication

Author: lebauce

analyzer/server.go

@@ -47,6 +47,7 @@ import (
 	usertopology "github.com/skydive-project/skydive/topology/enhancers"
 	"github.com/skydive-project/skydive/topology/probes/netlink"
 	"github.com/skydive-project/skydive/ui"
+	"github.com/skydive-project/skydive/websocket"
 	ws "github.com/skydive-project/skydive/websocket"
 )
 
@@ -251,8 +252,20 @@ func NewServerFromConfig() (*Server, error) {
 
 	uiServer.RegisterLoginRoute(apiAuthBackend)
 
+	peers, err := config.GetAnalyzerServiceAddresses()
+	if err != nil {
+		return nil, fmt.Errorf("Unable to get the analyzers list: %s", err)
+	}
+
+	opts := websocket.ServerOpts{
+		WriteCompression: true,
+		QueueSize:        10000,
+		PingDelay:        2 * time.Second,
+		PongTimeout:      5 * time.Second,
+	}
+
 	clusterAuthOptions := ClusterAuthenticationOpts()
-	hub, err := hub.NewHub(hserver, g, cached, apiAuthBackend, clusterAuthBackend, clusterAuthOptions, "/ws/agent/topology", true, 10000, 2*time.Second, 5*time.Second)
+	hub, err := hub.NewHub(hserver, g, cached, apiAuthBackend, clusterAuthBackend, clusterAuthOptions, "/ws/agent/topology", peers, opts)
 	if err != nil {
 		return nil, err
 	}

analyzer/storage.go

@@ -30,7 +30,7 @@ import (
 	es "github.com/skydive-project/skydive/storage/elasticsearch"
 )
 
-// NewESConfig returns a new elasticsearch configution for the given backend name
+// NewESConfig returns a new elasticsearch configuration for the given backend name
 func NewESConfig(name ...string) es.Config {
 	cfg := es.Config{}
 
@@ -63,13 +63,14 @@ func newGraphBackendFromConfig(etcdClient *etcd.Client) (graph.Backend, error) {
 		cfg := NewESConfig(backend)
 		return graph.NewElasticSearchBackendFromConfig(cfg, etcdClient)
 	case "memory":
-		return graph.NewMemoryBackend()
+		// cached memory will be used
+		return nil, nil
 	case "orientdb":
 		addr := config.GetString(configPath + ".addr")
 		database := config.GetString(configPath + ".database")
 		username := config.GetString(configPath + ".username")
 		password := config.GetString(configPath + ".password")
-		return graph.NewOrientDBBackend(addr, database, username, password)
+		return graph.NewOrientDBBackend(addr, database, username, password, etcdClient)
 	default:
 		return nil, fmt.Errorf("Topology backend driver '%s' not supported", driver)
 	}

cmd/client/topology.go

@@ -26,8 +26,8 @@ import (
 
 	"github.com/skydive-project/skydive/common"
 	"github.com/skydive-project/skydive/config"
-	"github.com/skydive-project/skydive/graffiti/graph"
 	"github.com/skydive-project/skydive/graffiti/hub"
+	gws "github.com/skydive-project/skydive/graffiti/websocket"
 	"github.com/skydive-project/skydive/websocket"
 	"github.com/spf13/cobra"
 )
@@ -91,7 +91,7 @@ var TopologyImport = &cobra.Command{
 			exitOnError(err)
 		}
 
-		syncMsg := []*graph.SyncMsg{}
+		syncMsg := []*gws.SyncMsg{}
 		if err := json.Unmarshal(content, &syncMsg); err != nil {
 			exitOnError(err)
 		}
@@ -101,14 +101,14 @@ var TopologyImport = &cobra.Command{
 		}
 
 		for _, node := range syncMsg[0].Nodes {
-			msg := websocket.NewStructMessage(graph.Namespace, graph.NodeAddedMsgType, node)
+			msg := gws.NewStructMessage(gws.NodeAddedMsgType, node)
 			if err := client.SendMessage(msg); err != nil {
 				exitOnError(fmt.Errorf("Failed to send message: %s", err))
 			}
 		}
 
 		for _, edge := range syncMsg[0].Edges {
-			msg := websocket.NewStructMessage(graph.Namespace, graph.EdgeAddedMsgType, edge)
+			msg := gws.NewStructMessage(gws.EdgeAddedMsgType, edge)
 			if err := client.SendMessage(msg); err != nil {
 				exitOnError(fmt.Errorf("Failed to send message: %s", err))
 			}

config/websocket.go

@@ -44,10 +44,14 @@ func NewWSClient(clientType common.ServiceType, url *url.URL, opts websocket.Cli
 
 // NewWSServer creates a Server based on the configuration
 func NewWSServer(server *shttp.Server, endpoint string, authBackend shttp.AuthenticationBackend) *websocket.Server {
-	queueSize := GetInt("http.ws.queue_size")
-	writeCompression := GetBool("http.ws.enable_write_compression")
 	pingDelay := time.Duration(GetInt("http.ws.ping_delay")) * time.Second
-	pongTimeout := time.Duration(GetInt("http.ws.pong_timeout"))*time.Second + pingDelay
 
-	return websocket.NewServer(server, endpoint, authBackend, writeCompression, queueSize, pingDelay, pongTimeout)
+	opts := websocket.ServerOpts{
+		WriteCompression: GetBool("http.ws.enable_write_compression"),
+		QueueSize:        GetInt("http.ws.queue_size"),
+		PingDelay:        pingDelay,
+		PongTimeout:      time.Duration(GetInt("http.ws.pong_timeout"))*time.Second + pingDelay,
+	}
+
+	return websocket.NewServer(server, endpoint, authBackend, opts)
 }

flow/storage/orientdb/orientdb.go

@@ -270,7 +270,7 @@ func (c *Storage) SearchRawPackets(fsq filters.SearchQuery, packetFilter *filter
 		}
 	}
 
-	result, err := c.client.Search(sql)
+	result, err := c.client.SQL(sql)
 	if err != nil {
 		return nil, err
 	}
@@ -316,7 +316,7 @@ func (c *Storage) SearchMetrics(fsq filters.SearchQuery, metricFilter *filters.F
 		}
 	}
 
-	result, err := c.client.Search(sql)
+	result, err := c.client.SQL(sql)
 	if err != nil {
 		return nil, err
 	}

graffiti/cmd/hub.go

@@ -31,6 +31,7 @@ import (
 	ge "github.com/skydive-project/skydive/gremlin/traversal"
 	shttp "github.com/skydive-project/skydive/http"
 	"github.com/skydive-project/skydive/logging"
+	"github.com/skydive-project/skydive/websocket"
 	"github.com/spf13/cobra"
 )
 
@@ -91,7 +92,7 @@ var HubCmd = &cobra.Command{
 
 		authBackend := shttp.NewNoAuthenticationBackend()
 
-		// declare all extension available throught API and filtering
+		// declare all extension available through API and filtering
 		tr := traversal.NewGremlinTraversalParser()
 		tr.AddTraversalExtension(ge.NewDescendantsTraversalExtension())
 
@@ -101,7 +102,14 @@ var HubCmd = &cobra.Command{
 		}
 		api.RegisterTopologyAPI(httpServer, g, tr, authBackend)
 
-		hub, err := hub.NewHub(httpServer, g, cached, authBackend, authBackend, nil, "/ws/pod", writeCompression, queueSize, time.Second*time.Duration(pingDelay), time.Second*time.Duration(pongTimeout))
+		serverOpts := websocket.ServerOpts{
+			WriteCompression: writeCompression,
+			QueueSize:        queueSize,
+			PingDelay:        time.Second * time.Duration(pingDelay),
+			PongTimeout:      time.Second * time.Duration(pongTimeout),
+		}
+
+		hub, err := hub.NewHub(httpServer, g, cached, authBackend, authBackend, nil, "/ws/pod", nil, serverOpts)
 		if err != nil {
 			logging.GetLogger().Error(err)
 			os.Exit(1)