// interpolateCounts compensates for sampling by artifically inflating counts
// throughout the report. It should be run once for each report, within the
// probe, before it gets emitted into the rest of the system.
func interpolateCounts(r report.Report) {
	rate := r.Sampling.Rate()
	if rate >= 1.0 {
		return
	}
	factor := 1.0 / rate
	for _, topology := range r.Topologies() {
		for _, nmd := range topology.Nodes {
			for _, emd := range nmd.Edges {
				if emd.EgressPacketCount != nil {
					*emd.EgressPacketCount = uint64(float64(*emd.EgressPacketCount) * factor)
				}
				if emd.IngressPacketCount != nil {
					*emd.IngressPacketCount = uint64(float64(*emd.IngressPacketCount) * factor)
				}
				if emd.EgressByteCount != nil {
					*emd.EgressByteCount = uint64(float64(*emd.EgressByteCount) * factor)
				}
				if emd.IngressByteCount != nil {
					*emd.IngressByteCount = uint64(float64(*emd.IngressByteCount) * factor)
				}
			}
		}
	}
}

func (r *Reporter) addConnection(rpt *report.Report, t fourTuple, extraFromNode, extraToNode map[string]string) {
	var (
		fromEndpointNodeID = report.MakeEndpointNodeID(r.hostID, t.fromAddr, strconv.Itoa(int(t.fromPort)))
		toEndpointNodeID   = report.MakeEndpointNodeID(r.hostID, t.toAddr, strconv.Itoa(int(t.toPort)))

		fromNode = report.MakeNodeWith(fromEndpointNodeID, map[string]string{
			Addr: t.fromAddr,
			Port: strconv.Itoa(int(t.fromPort)),
		}).WithEdge(toEndpointNodeID, report.EdgeMetadata{})
		toNode = report.MakeNodeWith(toEndpointNodeID, map[string]string{
			Addr: t.toAddr,
			Port: strconv.Itoa(int(t.toPort)),
		})
	)

	// In case we have a reverse resolution for the IP, we can use it for
	// the name...
	if toNames, err := r.reverseResolver.get(t.toAddr); err == nil {
		toNode = toNode.WithSet(ReverseDNSNames, report.MakeStringSet(toNames...))
	}

	if extraFromNode != nil {
		fromNode = fromNode.WithLatests(extraFromNode)
	}
	if extraToNode != nil {
		toNode = toNode.WithLatests(extraToNode)
	}
	rpt.Endpoint = rpt.Endpoint.AddNode(fromNode)
	rpt.Endpoint = rpt.Endpoint.AddNode(toNode)
}

// Tag implements Tagger.
func (t Tagger) Tag(r report.Report) (report.Report, error) {
	other := report.MakeNodeMetadataWith(map[string]string{report.HostNodeID: t.hostNodeID})
	for _, topology := range r.Topologies() {
		for id, md := range topology.NodeMetadatas {
			topology.NodeMetadatas[id] = md.Merge(other)
		}
	}
	return r, nil
}

func (t originHostTagger) Tag(r report.Report) (report.Report, error) {
	for _, topology := range r.Topologies() {
		md := report.NodeMetadata{report.HostNodeID: t.hostNodeID}
		for nodeID := range topology.NodeMetadatas {
			topology.NodeMetadatas[nodeID].Merge(md)
		}
	}
	return r, nil
}

// Tag implements Tagger.
func (t Tagger) Tag(r report.Report) (report.Report, error) {
	md := report.NewNodeMetadata(map[string]string{report.HostNodeID: t.hostNodeID})
	for _, topology := range r.Topologies() {
		for nodeID := range topology.NodeMetadatas {
			topology.NodeMetadatas[nodeID].Merge(md)
		}
	}
	return r, nil
}

// NodeTables produces a list of tables (to be consumed directly by the UI) based
// on the report and the node.  It uses the report to get the templates for the node's
// topology.
func NodeTables(r report.Report, n report.Node) []report.Table {
	if _, ok := n.Counters.Lookup(n.Topology); ok {
		// This is a group of nodes, so no tables!
		return nil
	}

	if topology, ok := r.Topology(n.Topology); ok {
		return topology.TableTemplates.Tables(n)
	}
	return nil
}

// NodeMetadata produces a table (to be consumed directly by the UI) based on
// an a report.Node, which is (hopefully) a node in one of our topologies.
func NodeMetadata(r report.Report, n report.Node) []report.MetadataRow {
	if _, ok := n.Counters.Lookup(n.Topology); ok {
		// This is a group of nodes, so no metadata!
		return nil
	}

	if topology, ok := r.Topology(n.Topology); ok {
		return topology.MetadataTemplates.MetadataRows(n)
	}
	return nil
}

// NodeMetrics produces a table (to be consumed directly by the UI) based on
// an a report.Node, which is (hopefully) a node in one of our topologies.
func NodeMetrics(r report.Report, n report.Node) []report.MetricRow {
	if _, ok := n.Counters.Lookup(n.Topology); ok {
		// This is a group of nodes, so no metrics!
		return nil
	}

	topology, ok := r.Topology(n.Topology)
	if !ok {
		return nil
	}
	return topology.MetricTemplates.MetricRows(n)
}

func (p *Probe) drainAndPublish(rpt report.Report, rs chan report.Report) {
ForLoop:
	for {
		select {
		case r := <-rs:
			rpt = rpt.Merge(r)
		default:
			break ForLoop
		}
	}

	if err := p.publisher.Publish(rpt); err != nil {
		log.Printf("publish: %v", err)
	}
}

func getRenderingContext(r report.Report, n RenderableNode) (multiContainer, multiHost bool) {
	var (
		originHosts      = map[string]struct{}{}
		originContainers = map[string]struct{}{}
	)
	for _, id := range n.Origins {
		for _, topology := range r.Topologies() {
			if nmd, ok := topology.Nodes[id]; ok {
				originHosts[report.ExtractHostID(nmd)] = struct{}{}
				if id, ok := nmd.Metadata[docker.ContainerID]; ok {
					originContainers[id] = struct{}{}
				}
			}
			// Return early if possible
			multiHost = len(originHosts) > 1
			multiContainer = len(originContainers) > 1
			if multiHost && multiContainer {
				return
			}
		}
	}
	return
}

func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr string, localPort, remotePort uint16, extraLocalNode, extraRemoteNode *report.Node) {
	localIsClient := int(localPort) > int(remotePort)

	// Update address topology
	{
		var (
			localAddressNodeID  = report.MakeAddressNodeID(r.hostID, localAddr)
			remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, remoteAddr)
			localNode           = report.MakeNodeWith(map[string]string{
				"name": r.hostName,
				Addr:   localAddr,
			})
			remoteNode = report.MakeNodeWith(map[string]string{
				Addr: remoteAddr,
			})
		)

		// In case we have a reverse resolution for the IP, we can use it for
		// the name...
		if revRemoteName, err := r.revResolver.Get(remoteAddr); err == nil {
			remoteNode = remoteNode.WithMetadata(map[string]string{
				"name": revRemoteName,
			})
		}

		if localIsClient {
			// New nodes are merged into the report so we don't need to do any
			// counting here; the merge does it for us.
			localNode = localNode.WithEdge(remoteAddressNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		} else {
			remoteNode = localNode.WithEdge(localAddressNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		}

		if extraLocalNode != nil {
			localNode = localNode.Merge(*extraLocalNode)
		}
		if extraRemoteNode != nil {
			remoteNode = remoteNode.Merge(*extraRemoteNode)
		}
		rpt.Address = rpt.Address.AddNode(localAddressNodeID, localNode)
		rpt.Address = rpt.Address.AddNode(remoteAddressNodeID, remoteNode)
	}

	// Update endpoint topology
	if r.includeProcesses {
		var (
			localEndpointNodeID  = report.MakeEndpointNodeID(r.hostID, localAddr, strconv.Itoa(int(localPort)))
			remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, remoteAddr, strconv.Itoa(int(remotePort)))

			localNode = report.MakeNodeWith(map[string]string{
				Addr: localAddr,
				Port: strconv.Itoa(int(localPort)),
			})
			remoteNode = report.MakeNodeWith(map[string]string{
				Addr: remoteAddr,
				Port: strconv.Itoa(int(remotePort)),
			})
		)

		// In case we have a reverse resolution for the IP, we can use it for
		// the name...
		if revRemoteName, err := r.revResolver.Get(remoteAddr); err == nil {
			remoteNode = remoteNode.WithMetadata(map[string]string{
				"name": revRemoteName,
			})
		}

		if localIsClient {
			// New nodes are merged into the report so we don't need to do any
			// counting here; the merge does it for us.
			localNode = localNode.WithEdge(remoteEndpointNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		} else {
			remoteNode = remoteNode.WithEdge(localEndpointNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		}

		if extraLocalNode != nil {
			localNode = localNode.Merge(*extraLocalNode)
		}
		if extraRemoteNode != nil {
			remoteNode = remoteNode.Merge(*extraRemoteNode)
		}
		rpt.Endpoint = rpt.Endpoint.AddNode(localEndpointNodeID, localNode)
		rpt.Endpoint = rpt.Endpoint.AddNode(remoteEndpointNodeID, remoteNode)
	}
}

// Publish serialises and compresses a report, then passes it to a publisher
func (p *ReportPublisher) Publish(r report.Report) error {
	buf := &bytes.Buffer{}
	r.WriteBinary(buf)
	return p.publisher.Publish(buf)
}

// Merge puts the packet into the report.
//
// Note that, for the moment, we encode bidirectional traffic as ingress and
// egress traffic on a single edge whose src is local and dst is remote. That
// is, if we see a packet from the remote addr 9.8.7.6 to the local addr
// 1.2.3.4, we apply it as *ingress* on the edge (1.2.3.4 -> 9.8.7.6).
func (s *Sniffer) Merge(p Packet, rpt *report.Report) {
	if p.SrcIP == "" || p.DstIP == "" {
		return
	}

	// One end of the traffic has to be local. Otherwise, we don't know how to
	// construct the edge.
	//
	// If we need to get around this limitation, we may be able to change the
	// semantics of the report, and allow the src side of edges to be from
	// anywhere. But that will have ramifications throughout Scope (read: it
	// may violate implicit invariants) and needs to be thought through.
	var (
		srcLocal   = s.localNets.Contains(net.ParseIP(p.SrcIP))
		dstLocal   = s.localNets.Contains(net.ParseIP(p.DstIP))
		localIP    string
		remoteIP   string
		localPort  string
		remotePort string
		egress     bool
	)
	switch {
	case srcLocal && !dstLocal:
		localIP, localPort, remoteIP, remotePort, egress = p.SrcIP, p.SrcPort, p.DstIP, p.DstPort, true
	case !srcLocal && dstLocal:
		localIP, localPort, remoteIP, remotePort, egress = p.DstIP, p.DstPort, p.SrcIP, p.SrcPort, false
	case srcLocal && dstLocal:
		localIP, localPort, remoteIP, remotePort, egress = p.SrcIP, p.SrcPort, p.DstIP, p.DstPort, true // loopback
	case !srcLocal && !dstLocal:
		log.Printf("sniffer ignoring remote-to-remote (%s -> %s) traffic", p.SrcIP, p.DstIP)
		return
	}

	addAdjacency := func(t report.Topology, srcNodeID, dstNodeID string) report.Topology {
		result := t.WithNode(srcNodeID, report.MakeNode().WithAdjacent(dstNodeID))
		result = result.WithNode(dstNodeID, report.MakeNode())
		return result
	}

	// For sure, we can add to the address topology.
	{
		var (
			srcNodeID = report.MakeAddressNodeID(s.hostID, localIP)
			dstNodeID = report.MakeAddressNodeID(s.hostID, remoteIP)
		)

		rpt.Address = addAdjacency(rpt.Address, srcNodeID, dstNodeID)

		emd := rpt.Address.Nodes[srcNodeID].Edges[dstNodeID]
		if egress {
			if emd.EgressPacketCount == nil {
				emd.EgressPacketCount = new(uint64)
			}
			*emd.EgressPacketCount++
			if emd.EgressByteCount == nil {
				emd.EgressByteCount = new(uint64)
			}
			*emd.EgressByteCount += uint64(p.Network)
		} else {
			if emd.IngressPacketCount == nil {
				emd.IngressPacketCount = new(uint64)
			}
			*emd.IngressPacketCount++
			if emd.IngressByteCount == nil {
				emd.IngressByteCount = new(uint64)
			}
			*emd.IngressByteCount += uint64(p.Network)
		}
		rpt.Address.Nodes[srcNodeID].Edges[dstNodeID] = emd
	}

	// If we have ports, we can add to the endpoint topology, too.
	if p.SrcPort != "" && p.DstPort != "" {
		var (
			srcNodeID = report.MakeEndpointNodeID(s.hostID, localIP, localPort)
			dstNodeID = report.MakeEndpointNodeID(s.hostID, remoteIP, remotePort)
		)

		rpt.Endpoint = addAdjacency(rpt.Endpoint, srcNodeID, dstNodeID)

		emd := rpt.Endpoint.Nodes[srcNodeID].Edges[dstNodeID]
		if egress {
			if emd.EgressPacketCount == nil {
				emd.EgressPacketCount = new(uint64)
			}
			*emd.EgressPacketCount++
			if emd.EgressByteCount == nil {
				emd.EgressByteCount = new(uint64)
			}
			*emd.EgressByteCount += uint64(p.Transport)
		} else {
			if emd.IngressPacketCount == nil {
				emd.IngressPacketCount = new(uint64)
			}
//.........这里部分代码省略.........

// Render implements Renderer
func (t TopologySelector) Render(r report.Report, _ Decorator) report.Nodes {
	topology, _ := r.Topology(string(t))
	return topology.Nodes
}

func controls(r report.Report, n report.Node) []ControlInstance {
	if t, ok := r.Topology(n.Topology); ok {
		return controlsFor(t, n.ID)
	}
	return []ControlInstance{}
}

func (r *Reporter) addConnection(rpt *report.Report, localAddr, remoteAddr string, localPort, remotePort uint16, proc *procspy.Proc) {
	var (
		localIsClient = int(localPort) > int(remotePort)
		hostNodeID    = report.MakeHostNodeID(r.hostID)
	)

	// Update address topology
	{
		var (
			localAddressNodeID  = report.MakeAddressNodeID(r.hostID, localAddr)
			remoteAddressNodeID = report.MakeAddressNodeID(r.hostID, remoteAddr)
			localNode           = report.MakeNodeWith(map[string]string{
				"name":            r.hostName,
				Addr:              localAddr,
				report.HostNodeID: hostNodeID,
			})
			remoteNode = report.MakeNodeWith(map[string]string{
				Addr: remoteAddr,
			})
		)

		if localIsClient {
			// New nodes are merged into the report so we don't need to do any counting here; the merge does it for us.
			localNode = localNode.WithEdge(remoteAddressNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		} else {
			remoteNode = localNode.WithEdge(localAddressNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		}

		rpt.Address = rpt.Address.WithNode(localAddressNodeID, localNode)
		rpt.Address = rpt.Address.WithNode(remoteAddressNodeID, remoteNode)
	}

	// Update endpoint topology
	if r.includeProcesses {
		var (
			localEndpointNodeID  = report.MakeEndpointNodeID(r.hostID, localAddr, strconv.Itoa(int(localPort)))
			remoteEndpointNodeID = report.MakeEndpointNodeID(r.hostID, remoteAddr, strconv.Itoa(int(remotePort)))

			localNode = report.MakeNodeWith(map[string]string{
				Addr:              localAddr,
				Port:              strconv.Itoa(int(localPort)),
				report.HostNodeID: hostNodeID,
			})
			remoteNode = report.MakeNodeWith(map[string]string{
				Addr: remoteAddr,
				Port: strconv.Itoa(int(remotePort)),
			})
		)

		if localIsClient {
			// New nodes are merged into the report so we don't need to do any counting here; the merge does it for us.
			localNode = localNode.WithEdge(remoteEndpointNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		} else {
			remoteNode = remoteNode.WithEdge(localEndpointNodeID, report.EdgeMetadata{
				MaxConnCountTCP: newu64(1),
			})
		}

		if proc != nil && proc.PID > 0 {
			localNode.Metadata[process.PID] = strconv.FormatUint(uint64(proc.PID), 10)
		}

		rpt.Endpoint = rpt.Endpoint.WithNode(localEndpointNodeID, localNode)
		rpt.Endpoint = rpt.Endpoint.WithNode(remoteEndpointNodeID, remoteNode)
	}
}