diff --git a/common/ip_resolver.go b/common/ip_resolver.go index 662f252a..db7ab496 100644 --- a/common/ip_resolver.go +++ b/common/ip_resolver.go @@ -48,6 +48,12 @@ type clusterSnapshot struct { PodDescriptors sync.Map // map[types.UID]Workload } +type InstanceMeta struct { + Name string + Region string + Zone string +} + type K8sIPResolver struct { clientset kubernetes.Interface snapshot clusterSnapshot @@ -416,6 +422,17 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { } for _, podIp := range pod.Status.PodIPs { resolver.storeWorkloadsIP(podIp.IP, &entry) + nodeInfo, ok := resolver.nodeInfoMap.Load(pod.Spec.NodeName) + region, zone := "", "" + if ok { + meta, ok := nodeInfo.(InstanceMeta) + if ok { + region = meta.Region + zone = meta.Zone + } else { + log.Println("Failed to fetch node info ", ok) + } + } podWorkload := Workload{ Name: pod.Name, Namespace: pod.Namespace, @@ -449,6 +466,17 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { } for _, podIp := range pod.Status.PodIPs { resolver.storeWorkloadsIP(podIp.IP, &entry) + nodeInfo, ok := resolver.nodeInfoMap.Load(pod.Spec.NodeName) + region, zone := "", "" + if ok { + meta, ok := nodeInfo.(InstanceMeta) + if ok { + region = meta.Region + zone = meta.Zone + } + } else { + log.Println("Failed to fetch node info ", ok) + } podWorkload := Workload{ Name: pod.Name, Namespace: pod.Namespace, @@ -457,6 +485,7 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { Zone: zone, Instance: pod.Spec.NodeName, } + resolver.storePodsIP(podIp.IP, &podWorkload) } case watch.Deleted: @@ -494,6 +523,8 @@ func (resolver *K8sIPResolver) handleNodeWatchEvent(nodeEvent *watch.Event) { Instance: node.Name, }) } + nodeMetadata := InstanceMeta{Zone: node.Annotations["topology.kubernetes.io/zone"], Region: node.Annotations["topology.kubernetes.io/region"]} + resolver.nodeInfoMap.Store(node.Name, nodeMetadata) case watch.Deleted: if val, ok := nodeEvent.Object.(*v1.Node); ok { resolver.snapshot.Nodes.Delete(val.UID) diff --git a/containers/container.go b/containers/container.go index 8ecfd0b1..67fe0425 100644 --- a/containers/container.go +++ b/containers/container.go @@ -158,7 +158,6 @@ type Container struct { done chan struct{} ip_resolver IPResolver - hostIpsMap sync.Map } func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, hostConntrack *Conntrack, pid uint32, registry *Registry) (*Container, error) { @@ -329,8 +328,9 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { } if d.dstWorkload.Kind == "external" { - if host, ok := c.hostIpsMap.Load(d.dst); ok { - workload_dest = common.Workload{Name: host.(string), Namespace: "external", Kind: "external"} + host := c.ip_resolver.ResolveHost(d.dst.IP().String()) + if host != "" { + workload_dest = common.Workload{Name: host, Namespace: "external", Kind: "external"} } } ch <- counter(metrics.NetConnectionsSuccessful, float64(stats.Count), d.src.String(), d.dst.String(), workload_src.Name, workload_src.Namespace, workload_src.Kind, workload_dest.Name, workload_dest.Namespace, workload_dest.Kind, actualDestWorkload.Name, actualDestWorkload.Namespace, actualDestWorkload.Kind) @@ -355,8 +355,9 @@ func (c *Container) Collect(ch chan<- prometheus.Metric) { workload_dest := c.ip_resolver.ResolveIP(conn.ActualDest.IP().String()) actualDestWorkload := c.ip_resolver.ResolveActualIP(conn.ActualDest.IP().String()) if workload_dest.Kind == "external" { - if host, ok := c.hostIpsMap.Load(conn.ActualDest.IP().String()); ok { - workload_dest = common.Workload{Name: host.(string), Namespace: "external", Kind: "external"} + host := c.ip_resolver.ResolveHost(conn.ActualDest.IP().String()) + if host != "" { + workload_dest = common.Workload{Name: host, Namespace: "external", Kind: "external"} } } connections[AddrPair{src: addrPair.dst, dst: conn.ActualDest, srcWorkload: workload_src, dstWorkload: workload_dest, actualDestWorkload: actualDestWorkload}]++ @@ -815,7 +816,7 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R if r.Response != nil { response = base64.StdEncoding.EncodeToString(r.Response) } - trace.HttpRequest(method, uri, r.Status, r.Duration, r.PayloadSize, payload, headers, response, host) + trace.HttpRequest(method, uri, r.Status, r.Duration, r.PayloadSize, payload, headers, response, host, conn.actualDestWorkload) case l7.ProtocolHTTP2: if conn.http2Parser == nil { conn.http2Parser = l7.NewHttp2Parser() diff --git a/containers/registry.go b/containers/registry.go index 2598892f..730e1419 100644 --- a/containers/registry.go +++ b/containers/registry.go @@ -38,7 +38,8 @@ type ProcessInfo struct { type IPResolver interface { ResolveIP(string) common.Workload ResolveActualIP(string) common.Workload - CacheDNS(string, string) common.Workload + CacheDNS(string, string) + ResolveHost(string) string StartWatching() error StopWatching() } diff --git a/main.go b/main.go index 9ea2da0a..2607e178 100644 --- a/main.go +++ b/main.go @@ -167,7 +167,7 @@ func main() { registerer.MustRegister(info("node_agent_info", version)) - if err := registerer.Register(node.NewCollector(hostname, kv)); err != nil { + if err := registerer.Register(node.NewCollector(hostname, kv, md)); err != nil { klog.Exitln(err) } diff --git a/node/collector.go b/node/collector.go index fc86458f..d2358be4 100644 --- a/node/collector.go +++ b/node/collector.go @@ -152,8 +152,7 @@ type Collector struct { instanceMetadata *metadata.CloudMetadata } -func NewCollector(hostname, kernelVersion string) *Collector { - md := metadata.GetInstanceMetadata() +func NewCollector(hostname, kernelVersion string, md *metadata.CloudMetadata) *Collector { klog.Infof("instance metadata: %+v", md) return &Collector{ hostname: hostname, diff --git a/tracing/tracing.go b/tracing/tracing.go index d1874c7e..d05fecf3 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -27,12 +27,14 @@ const ( ) var ( - tracer func(containerId string) trace.Tracer + tracer func(containerId string) trace.Tracer + instanceMetadata *metadata.CloudMetadata ) func Init(machineId, hostname, version string) { md := metadata.GetInstanceMetadata() endpointUrl := *flags.TracesEndpoint + instanceMetadata = md if endpointUrl == nil { klog.Infoln("no OpenTelemetry traces collector endpoint configured") return @@ -144,7 +146,7 @@ func (t *Trace) createSpan(name string, duration time.Duration, error bool, attr span.End(trace.WithTimestamp(end)) } -func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time.Duration, requestSize uint64, payload string, headers string, response string, host string) { +func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time.Duration, requestSize uint64, payload string, headers string, response string, host string, destWorkload common.Workload) { if t == nil || method == "" { return }