From 79804a5f511660c64dcdcab80d38d8a07d469f2e Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Mon, 6 May 2024 07:33:16 +0000 Subject: [PATCH] feat: add network az and region info --- common/ip_resolver.go | 40 ++++++++++++++++++++++++++++++++++++++++ containers/container.go | 20 +++++++++++--------- tracing/tracing.go | 12 +++++++++++- 3 files changed, 62 insertions(+), 10 deletions(-) diff --git a/common/ip_resolver.go b/common/ip_resolver.go index c7b1714d..c7b89955 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 @@ -56,12 +62,15 @@ type K8sIPResolver struct { shouldResolveDns bool dnsResolvedIps *lrucache.Cache[string, string] podIpsMap sync.Map + nodeInfoMap sync.Map } type Workload struct { Name string Namespace string Kind string + Zone string + Region string } func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPResolver, error) { @@ -83,6 +92,7 @@ func NewK8sIPResolver(clientset kubernetes.Interface, resolveDns bool) (*K8sIPRe shouldResolveDns: resolveDns, dnsResolvedIps: dnsCache, podIpsMap: sync.Map{}, + nodeInfoMap: sync.Map{}, }, nil } @@ -380,10 +390,21 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { entry := resolver.resolvePodDescriptor(pod) for _, podIp := range pod.Status.PodIPs { resolver.storeWorkloadsIP(podIp.IP, &entry) + nodeInfo, err := resolver.nodeInfoMap.Load(pod.Spec.NodeName) + region, zone := "", "" + if !err { + meta, ok := nodeInfo.(InstanceMeta) + if ok { + region = meta.Region + zone = meta.Zone + } + } podWorkload := Workload{ Name: pod.Name, Namespace: pod.Namespace, Kind: "pod", + Region: region, + Zone: zone, } resolver.storePodsIP(podIp.IP, &podWorkload) } @@ -396,11 +417,23 @@ func (resolver *K8sIPResolver) handlePodWatchEvent(podEvent *watch.Event) { entry := resolver.resolvePodDescriptor(pod) for _, podIp := range pod.Status.PodIPs { resolver.storeWorkloadsIP(podIp.IP, &entry) + nodeInfo, err := resolver.nodeInfoMap.Load(pod.Spec.NodeName) + region, zone := "", "" + if !err { + meta, ok := nodeInfo.(InstanceMeta) + if ok { + region = meta.Region + zone = meta.Zone + } + } podWorkload := Workload{ Name: pod.Name, Namespace: pod.Namespace, Kind: "pod", + Region: region, + Zone: zone, } + resolver.storePodsIP(podIp.IP, &podWorkload) } case watch.Deleted: @@ -424,11 +457,16 @@ func (resolver *K8sIPResolver) handleNodeWatchEvent(nodeEvent *watch.Event) { Name: node.Name, Namespace: "node", Kind: "node", + Zone: node.Annotations["topology.kubernetes.io/zone"], + Region: node.Annotations["topology.kubernetes.io/region"], }) } + 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) + resolver.nodeInfoMap.Delete(val.Name) } } } @@ -499,6 +537,8 @@ func (resolver *K8sIPResolver) handleServicesWatchEvent(servicesEvent *watch.Eve Name: service.Name, Namespace: service.Namespace, Kind: "Service", + Zone: "", + Region: "", } // TODO maybe try to match service to workload diff --git a/containers/container.go b/containers/container.go index 83252b79..f1f85f89 100644 --- a/containers/container.go +++ b/containers/container.go @@ -17,6 +17,7 @@ import ( "github.com/coroot/coroot-node-agent/flags" "github.com/coroot/coroot-node-agent/logs" "github.com/coroot/coroot-node-agent/node" + "github.com/coroot/coroot-node-agent/node/metadata" "github.com/coroot/coroot-node-agent/pinger" "github.com/coroot/coroot-node-agent/proc" "github.com/coroot/coroot-node-agent/tracing" @@ -140,12 +141,14 @@ type Container struct { lock sync.RWMutex - done chan struct{} - ip_resolver IPResolver - hostIpsMap sync.Map + done chan struct{} + ip_resolver IPResolver + hostIpsMap sync.Map + instanceMetadata *metadata.CloudMetadata } func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, hostConntrack *Conntrack, pid uint32, ip_resolver IPResolver) (*Container, error) { + instanceMetadata := metadata.GetInstanceMetadata() netNs, err := proc.GetNetNs(pid) if err != nil { return nil, err @@ -177,9 +180,10 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, host hostConntrack: hostConntrack, - done: make(chan struct{}), - ip_resolver: ip_resolver, - hostIpsMap: sync.Map{}, + done: make(chan struct{}), + ip_resolver: ip_resolver, + hostIpsMap: sync.Map{}, + instanceMetadata: instanceMetadata, } for _, n := range md.networks { @@ -551,14 +555,12 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP } srcWorkload := c.ip_resolver.ResolveIP(src.IP().String()) if ignoreControlPlane(srcWorkload.Name) { - klog.Warningf("Ignoring src workload %s, %s \n", src.IP().String(), srcWorkload.Name) return } actualDst, err := c.getActualDestination(p, src, dst) dstWorkload := c.ip_resolver.ResolveIP(dst.IP().String()) if ignoreControlPlane(dstWorkload.Name) { - klog.Warningf("Ignoring src workload %s, %s \n", dst.IP().String(), dstWorkload.Name) return } if err != nil { @@ -693,7 +695,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, *c.instanceMetadata, conn.actualDestWorkload) case l7.ProtocolHTTP2: if conn.http2Parser == nil { conn.http2Parser = l7.NewHttp2Parser() diff --git a/tracing/tracing.go b/tracing/tracing.go index dac1ca5b..3a3d4d7f 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -8,6 +8,7 @@ import ( "github.com/coroot/coroot-node-agent/common" "github.com/coroot/coroot-node-agent/ebpftracer/l7" "github.com/coroot/coroot-node-agent/flags" + "github.com/coroot/coroot-node-agent/node/metadata" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/exporters/otlp/otlptrace" @@ -107,7 +108,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, instanceMetadata metadata.CloudMetadata, destWorkload common.Workload) { if t == nil || method == "" { return } @@ -123,6 +124,15 @@ func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time attribute.Key("http.headers").String(headers), attribute.Key("http.response").String(response), attribute.Key("http.path").String(path), + attribute.Key("source.cloud.region").String(instanceMetadata.Region), + attribute.Key("source.cloud.provider").String(string(instanceMetadata.Provider)), + attribute.Key("source.cloud.zone").String(instanceMetadata.AvailabilityZone), + attribute.Key("source.cloud.zone_id").String(instanceMetadata.AvailabilityZoneId), + attribute.Key("source.cloud.instance_id").String(instanceMetadata.InstanceId), + attribute.Key("source.cloud.instance_type").String(instanceMetadata.InstanceType), + attribute.Key("source.cloud.region").String(instanceMetadata.Region), + attribute.Key("destination.cloud.region").String(destWorkload.Region), + attribute.Key("destination.cloud.zone").String(destWorkload.Zone), ) }