|
| 1 | +package traversal |
| 2 | + |
| 3 | +import ( |
| 4 | + "github.com/pkg/errors" |
| 5 | + |
| 6 | + "github.com/skydive-project/skydive/graffiti/filters" |
| 7 | + "github.com/skydive-project/skydive/graffiti/graph" |
| 8 | + "github.com/skydive-project/skydive/graffiti/graph/traversal" |
| 9 | + "github.com/skydive-project/skydive/topology" |
| 10 | +) |
| 11 | + |
| 12 | +// NeighborsTraversalExtension describes a new extension to enhance the topology |
| 13 | +type NeighborsTraversalExtension struct { |
| 14 | + NeighborsToken traversal.Token |
| 15 | +} |
| 16 | + |
| 17 | +// NeighborsGremlinTraversalStep navigate the graph starting from a node, following edges |
| 18 | +// from parent to child and from child to parent. |
| 19 | +// It follows the same sintaxis as Ascendants and Descendants step. |
| 20 | +// The behaviour is like Ascendants+Descendants combined. |
| 21 | +// If only one param is defined, it is used as depth, eg: G.V('id').Neighbors(4) |
| 22 | +// If we have an event number of parameters, they are used as edge filter, and |
| 23 | +// depth is defaulted to one, eg.: G.V('id').Neighbors("Type","foo","RelationType","bar") |
| 24 | +// If we have an odd, but >1, number of parameters, all but the last one are used as |
| 25 | +// edge filters and the last one as depth, eg.: G.V('id').Neighbors("Type","foo","RelationType","bar",3) |
| 26 | +type NeighborsGremlinTraversalStep struct { |
| 27 | + context traversal.GremlinTraversalContext |
| 28 | + maxDepth int64 |
| 29 | + edgeFilter graph.ElementMatcher |
| 30 | + // nextStepOnlyIDs is set to true if the next step only needs node IDs and not the whole node info |
| 31 | + nextStepOnlyIDs bool |
| 32 | +} |
| 33 | + |
| 34 | +// NewNeighborsTraversalExtension returns a new graph traversal extension |
| 35 | +func NewNeighborsTraversalExtension() *NeighborsTraversalExtension { |
| 36 | + return &NeighborsTraversalExtension{ |
| 37 | + NeighborsToken: traversalNeighborsToken, |
| 38 | + } |
| 39 | +} |
| 40 | + |
| 41 | +// ScanIdent returns an associated graph token |
| 42 | +func (e *NeighborsTraversalExtension) ScanIdent(s string) (traversal.Token, bool) { |
| 43 | + switch s { |
| 44 | + case "NEIGHBORS": |
| 45 | + return e.NeighborsToken, true |
| 46 | + } |
| 47 | + return traversal.IDENT, false |
| 48 | +} |
| 49 | + |
| 50 | +// ParseStep parses neighbors step |
| 51 | +func (e *NeighborsTraversalExtension) ParseStep(t traversal.Token, p traversal.GremlinTraversalContext) (traversal.GremlinTraversalStep, error) { |
| 52 | + switch t { |
| 53 | + case e.NeighborsToken: |
| 54 | + default: |
| 55 | + return nil, nil |
| 56 | + } |
| 57 | + |
| 58 | + maxDepth := int64(1) |
| 59 | + edgeFilter, _ := topology.OwnershipMetadata().Filter() |
| 60 | + |
| 61 | + switch len(p.Params) { |
| 62 | + case 0: |
| 63 | + default: |
| 64 | + i := len(p.Params) / 2 * 2 |
| 65 | + filter, err := traversal.ParamsToFilter(filters.BoolFilterOp_OR, p.Params[:i]...) |
| 66 | + if err != nil { |
| 67 | + return nil, errors.Wrap(err, "Neighbors accepts an optional number of key/value tuples and an optional depth") |
| 68 | + } |
| 69 | + edgeFilter = filter |
| 70 | + |
| 71 | + if i == len(p.Params) { |
| 72 | + break |
| 73 | + } |
| 74 | + |
| 75 | + fallthrough |
| 76 | + case 1: |
| 77 | + depth, ok := p.Params[len(p.Params)-1].(int64) |
| 78 | + if !ok { |
| 79 | + return nil, errors.New("Neighbors last argument must be the maximum depth specified as an integer") |
| 80 | + } |
| 81 | + maxDepth = depth |
| 82 | + } |
| 83 | + |
| 84 | + return &NeighborsGremlinTraversalStep{context: p, maxDepth: maxDepth, edgeFilter: graph.NewElementFilter(edgeFilter)}, nil |
| 85 | +} |
| 86 | + |
| 87 | +// getNeighbors given a list of nodes, get its neighbors nodes for "maxDepth" depth relationships. |
| 88 | +// Edges between nodes must fulfill "edgeFilter" filter. |
| 89 | +// Nodes passed to this function will always be in the response. |
| 90 | +func (d *NeighborsGremlinTraversalStep) getNeighbors(g *graph.Graph, nodes []*graph.Node) []*graph.Node { |
| 91 | + // visitedNodes store neighors and avoid visiting twice the same node |
| 92 | + visitedNodes := map[graph.Identifier]interface{}{} |
| 93 | + |
| 94 | + // currentDepthNodesIDs slice with the nodes being processed in each depth. |
| 95 | + // We use "empty" while procesing the neighbors nodes to avoid extra calls to the backend. |
| 96 | + var currentDepthNodesIDs []graph.Identifier |
| 97 | + // nextDepthNodes slice were next depth nodes are being stored. |
| 98 | + // Initializated with the list of origin nodes where it should start from. |
| 99 | + nextDepthNodesIDs := make([]graph.Identifier, 0, len(nodes)) |
| 100 | + |
| 101 | + // Mark origin nodes as already visited |
| 102 | + // Neighbor step will return also the origin nodes |
| 103 | + for _, n := range nodes { |
| 104 | + visitedNodes[n.ID] = struct{}{} |
| 105 | + nextDepthNodesIDs = append(nextDepthNodesIDs, n.ID) |
| 106 | + } |
| 107 | + |
| 108 | + // DFS |
| 109 | + // BFS must not be used because could lead to ignore some servers in this case: |
| 110 | + // A -> B |
| 111 | + // B -> C |
| 112 | + // C -> D |
| 113 | + // A -> C |
| 114 | + // With depth=2, BFS will return A,B,C (C is visited in A->B->C, si ignored in A->C->D) |
| 115 | + // DFS will return, the correct, A,B,C,D |
| 116 | + for i := 0; i < int(d.maxDepth); i++ { |
| 117 | + // Copy values from nextDepthNodes to currentDepthNodes |
| 118 | + currentDepthNodesIDs = make([]graph.Identifier, len(nextDepthNodesIDs)) |
| 119 | + copy(currentDepthNodesIDs, nextDepthNodesIDs) |
| 120 | + |
| 121 | + nextDepthNodesIDs = nextDepthNodesIDs[:0] // Clean slice, keeping capacity |
| 122 | + // Get all edges for the list of nodes, filtered by edgeFilter |
| 123 | + // Convert the list of node ids to a list of nodes |
| 124 | + |
| 125 | + currentDepthNodes := make([]*graph.Node, 0, len(currentDepthNodesIDs)) |
| 126 | + for _, nID := range currentDepthNodesIDs { |
| 127 | + currentDepthNodes = append(currentDepthNodes, graph.CreateNode(nID, graph.Metadata{}, graph.Unix(0, 0), "", "")) |
| 128 | + } |
| 129 | + edges := g.GetNodesEdges(currentDepthNodes, d.edgeFilter) |
| 130 | + |
| 131 | + for _, e := range edges { |
| 132 | + // Get nodeID of the other side of the edge |
| 133 | + // Store neighbors |
| 134 | + // We don't know in which side of the edge are the neighbors, so, add both sides if not already visited |
| 135 | + _, okParent := visitedNodes[e.Parent] |
| 136 | + if !okParent { |
| 137 | + visitedNodes[e.Parent] = struct{}{} |
| 138 | + // Do not walk nodes already processed |
| 139 | + nextDepthNodesIDs = append(nextDepthNodesIDs, e.Parent) |
| 140 | + } |
| 141 | + _, okChild := visitedNodes[e.Child] |
| 142 | + if !okChild { |
| 143 | + visitedNodes[e.Child] = struct{}{} |
| 144 | + nextDepthNodesIDs = append(nextDepthNodesIDs, e.Child) |
| 145 | + } |
| 146 | + } |
| 147 | + } |
| 148 | + |
| 149 | + // Return "empty" nodes (just with the ID) if the next step only need that info |
| 150 | + if d.nextStepOnlyIDs { |
| 151 | + ret := make([]*graph.Node, 0, len(visitedNodes)) |
| 152 | + for nID := range visitedNodes { |
| 153 | + ret = append(ret, graph.CreateNode(nID, graph.Metadata{}, graph.Unix(0, 0), "", "")) |
| 154 | + } |
| 155 | + return ret |
| 156 | + } |
| 157 | + |
| 158 | + // Get concurrentl all nodes for the list of neighbors ids |
| 159 | + nodesIDs := make([]graph.Identifier, 0, len(visitedNodes)) |
| 160 | + for n := range visitedNodes { |
| 161 | + nodesIDs = append(nodesIDs, n) |
| 162 | + } |
| 163 | + |
| 164 | + return g.GetNodesFromIDs(nodesIDs) |
| 165 | +} |
| 166 | + |
| 167 | +// Exec Neighbors step |
| 168 | +func (d *NeighborsGremlinTraversalStep) Exec(last traversal.GraphTraversalStep) (traversal.GraphTraversalStep, error) { |
| 169 | + switch tv := last.(type) { |
| 170 | + case *traversal.GraphTraversalV: |
| 171 | + tv.GraphTraversal.RLock() |
| 172 | + neighbors := d.getNeighbors(tv.GraphTraversal.Graph, tv.GetNodes()) |
| 173 | + tv.GraphTraversal.RUnlock() |
| 174 | + |
| 175 | + return traversal.NewGraphTraversalV(tv.GraphTraversal, neighbors), nil |
| 176 | + } |
| 177 | + return nil, traversal.ErrExecutionError |
| 178 | +} |
| 179 | + |
| 180 | +// Reduce Neighbors step |
| 181 | +func (d *NeighborsGremlinTraversalStep) Reduce(next traversal.GremlinTraversalStep) (traversal.GremlinTraversalStep, error) { |
| 182 | + return next, nil |
| 183 | +} |
| 184 | + |
| 185 | +// Context Neighbors step |
| 186 | +func (d *NeighborsGremlinTraversalStep) Context() *traversal.GremlinTraversalContext { |
| 187 | + return &d.context |
| 188 | +} |
0 commit comments